Google Pub/Sub

# Criar topic
gcloud pubsub topics create my-topic

# Criar subscription
gcloud pubsub subscriptions create my-subscription --topic=my-topic

# Listar topics
gcloud pubsub topics list

# Listar subscriptions
gcloud pubsub subscriptions list

from google.cloud import pubsub_v1
import json
import time
from datetime import datetime

# Configurar cliente
project_id = "my-project"
topic_id = "my-topic"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

def publish_message(data):
    # Converter dados para bytes
    message_data = json.dumps(data).encode('utf-8')
    
    # Adicionar atributos
    attributes = {
        'timestamp': datetime.now().isoformat(),
        'source': 'python-publisher'
    }
    
    # Publicar mensagem
    future = publisher.publish(topic_path, message_data, **attributes)
    message_id = future.result()
    print(f"Published message ID: {message_id}")
    return message_id

# Exemplo de uso
for i in range(100):
    data = {
        'user_id': f'user_{i}',
        'event': 'page_view',
        'page': f'/page_{i % 10}',
        'timestamp': datetime.now().isoformat()
    }
    
    publish_message(data)
    time.sleep(0.1)

from google.cloud import pubsub_v1
import json
from concurrent.futures import ThreadPoolExecutor

project_id = "my-project"
subscription_id = "my-subscription"

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message):
    try:
        # Processar mensagem
        data = json.loads(message.data.decode('utf-8'))
        attributes = dict(message.attributes)
        
        print(f"Received message: {data}")
        print(f"Attributes: {attributes}")
        
        # Simular processamento
        time.sleep(0.1)
        
        # Acknowledge mensagem
        message.ack()
        print(f"Message acknowledged: {message.message_id}")
        
    except Exception as e:
        print(f"Error processing message: {e}")
        message.nack()

# Configurar subscriber
flow_control = pubsub_v1.types.FlowControl(max_messages=100)
executor = ThreadPoolExecutor(max_workers=10)

print(f"Listening for messages on {subscription_path}...")

# Iniciar subscriber
streaming_pull_future = subscriber.subscribe(
    subscription_path, 
    callback=callback,
    flow_control=flow_control,
    executor=executor
)

try:
    streaming_pull_future.result()
except KeyboardInterrupt:
    streaming_pull_future.cancel()
    print("Subscriber stopped.")

import json
import base64
from google.cloud import bigquery

def pubsub_trigger(event, context):
    """Cloud Function triggered by Pub/Sub message"""
    
    # Decodificar mensagem
    if 'data' in event:
        message_data = base64.b64decode(event['data']).decode('utf-8')
        data = json.loads(message_data)
    else:
        print('No data in Pub/Sub message')
        return
    
    # Obter atributos
    attributes = event.get('attributes', {})
    
    print(f"Processing message: {data}")
    print(f"Attributes: {attributes}")
    
    # Exemplo: inserir no BigQuery
    client = bigquery.Client()
    table_id = "my-project.my_dataset.events"
    
    rows_to_insert = [{
        'user_id': data.get('user_id'),
        'event': data.get('event'),
        'page': data.get('page'),
        'timestamp': data.get('timestamp'),
        'processed_at': datetime.utcnow().isoformat()
    }]
    
    errors = client.insert_rows_json(table_id, rows_to_insert)
    
    if errors:
        print(f"BigQuery insert errors: {errors}")
        raise Exception("Failed to insert into BigQuery")
    else:
        print("Data inserted successfully into BigQuery")

# Pub/Sub Topic
resource "google_pubsub_topic" "example" {
  name = "my-topic"
  
  labels = {
    environment = "production"
    team        = "data-engineering"
  }
  
  message_retention_duration = "86400s"
}

# Pub/Sub Subscription
resource "google_pubsub_subscription" "example" {
  name  = "my-subscription"
  topic = google_pubsub_topic.example.name
  
  # Configurações de entrega
  ack_deadline_seconds = 20
  
  # Dead letter policy
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.dead_letter.id
    max_delivery_attempts = 5
  }
  
  # Retry policy
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "600s"
  }
  
  # Push configuration (opcional)
  push_config {
    push_endpoint = "https://my-app.com/webhook"
    
    attributes = {
      x-goog-version = "v1"
    }
  }
}

# Dead Letter Topic
resource "google_pubsub_topic" "dead_letter" {
  name = "my-topic-dead-letter"
}