Google Pub/Sub #

gem install google-cloud-pubsub

# Or in the Gemfile

# Gemfile
gem 'google-cloud-pubsub', '~> 2.18'

require 'google/cloud/pubsub'

# Method 1: Application Default Credentials (ADC) — automatic from the environment
# Locally: run `gcloud auth application-default login`
# On GCE/GKE/Cloud Run: automatic from the attached service account
pubsub = Google::Cloud::PubSub.new(project_id: "my-project-id")

# Method 2: Service Account JSON file
pubsub = Google::Cloud::PubSub.new(
  project_id:  "my-project-id",
  credentials: "/path/to/service-account.json"
)

# Method 3: Environment variables (most common in production)
# GOOGLE_CLOUD_PROJECT=my-project-id
# GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
pubsub = Google::Cloud::PubSub.new

# Method 4: Direct credential hash (avoid in production)
pubsub = Google::Cloud::PubSub.new(
  project_id:  "my-project-id",
  credentials: {
    type:                 "service_account",
    project_id:           "my-project-id",
    private_key_id:       ENV["GCP_PRIVATE_KEY_ID"],
    private_key:          ENV["GCP_PRIVATE_KEY"],
    client_email:         ENV["GCP_CLIENT_EMAIL"],
    client_id:            ENV["GCP_CLIENT_ID"],
    auth_uri:             "https://accounts.google.com/o/oauth2/auth",
    token_uri:            "https://oauth2.googleapis.com/token"
  }
)

# Method 5: Emulator for local development (no connection to GCP)
# Run: gcloud beta emulators pubsub start
ENV["PUBSUB_EMULATOR_HOST"] = "localhost:8085"
pubsub = Google::Cloud::PubSub.new(project_id: "test-project")

pubsub = Google::Cloud::PubSub.new(project_id: "my-project")

# Create a topic
topic = pubsub.create_topic("orders")

# Or fetch an existing topic
topic = pubsub.topic("orders")

# Create a topic with configuration
topic = pubsub.create_topic(
  "transactions",
  message_retention_duration: "604800s",  # 7 day message retention
  kms_key_name: "projects/my-project/locations/global/keyRings/my-ring/cryptoKeys/my-key"
)

# List all topics
pubsub.topics.each { |t| puts t.name }

# Delete a topic
topic.delete

# ===== Subscriptions =====

# Pull Subscription — a consumer that actively requests messages
sub = topic.subscribe(
  "orders-notifications",
  deadline:                    60,     # ack deadline in seconds (max 600)
  retain_acked_messages:       false,  # true for replay capability
  message_retention_duration:  "604800s",  # 7 days
  enable_message_ordering:     false   # true for ordered delivery
)

# Push Subscription — Pub/Sub sends an HTTP POST to the endpoint
push_sub = topic.subscribe(
  "orders-webhook",
  endpoint: "https://api.example.com/webhook/pubsub"
)

# Subscription with a filter — only receive messages with certain attributes
filter_sub = topic.subscribe(
  "premium-orders",
  filter: 'attributes.customer_type = "premium"'
)

# Subscription with a Dead Letter Topic
dlq_topic = pubsub.create_topic("orders-dlq")
sub_with_dlq = topic.subscribe(
  "orders-with-dlq",
  dead_letter_topic:              dlq_topic,
  max_delivery_attempts:         5   # after 5 failures → DLQ
)

# Subscription with a retry policy
sub_with_retry = topic.subscribe(
  "orders-retry",
  minimum_backoff: 10,   # min 10 seconds before retry
  maximum_backoff: 600   # max 600 seconds (10 minutes)
)

# Fetch an existing subscription
sub = pubsub.subscription("orders-notifications")

# List subscriptions
topic.subscriptions.each { |s| puts s.name }

topic = pubsub.topic("orders")

# Publish a simple message
message_id = topic.publish(
  JSON.generate({
    event:     "order_created",
    order_id:  12345,
    total:     150_000,
    timestamp: Time.now.iso8601
  })
)
puts "Message ID: #{message_id}"

# Publish with attributes — usable for subscription filters
message_id = topic.publish(
  JSON.generate({ order_id: 12346, total: 250_000 }),
  "event_type"      => "new_order",
  "customer_type"   => "premium",
  "schema_version"  => "v2",
  "environment"     => Rails.env
)

# Publish with an ordering key — messages with the same key are delivered in order
# (the subscription must have enable_message_ordering: true)
topic.publish(
  JSON.generate({ order_id: 100, status: "created" }),
  ordering_key: "user-#{user_id}"
)

topic.publish(
  JSON.generate({ order_id: 100, status: "paid" }),
  ordering_key: "user-#{user_id}"
  # This message is guaranteed to be delivered AFTER the one above for the same key
)

# publish with a block — Pub/Sub batches automatically
topic.publish do |batch|
  order_list.each do |order|
    batch.publish(
      JSON.generate({
        order_id: order.id,
        total:    order.total,
        status:   order.status
      }),
      "event_type" => "order_sync"
    )
  end
end

# Async publisher — publishes in a background thread, non-blocking
publisher = topic.async_publisher(
  max_bytes:    1_000_000,   # flush when the buffer reaches 1MB
  max_messages: 100,         # flush when the buffer has 100 messages
  interval:     0.25         # flush every 250ms even if not full
)

# Callback for confirmations or errors
publisher.on_error do |exception|
  Rails.logger.error "Publish error: #{exception.message}"
end

# Async publish — returns immediately without waiting for confirmation
1000.times do |i|
  publisher.publish(
    JSON.generate({ seq: i, ts: Time.now.to_i }),
    "batch_id" => "batch-#{Time.now.to_i}"
  )
end

# Flush all messages still in the buffer
publisher.flush
publisher.stop.wait!   # wait for everything to finish before shutdown

sub = pubsub.subscription("orders-notifications")

# Streaming pull — the subscriber runs in background threads
# Far more efficient than manual polling
subscriber = sub.listen(
  streams:           4,    # number of parallel streaming connections
  inventory:         1000, # number of messages buffered
  threads: {
    callback: 8,   # threads for running callbacks
    push:     4    # threads for pushing acks/nacks to Pub/Sub
  }
) do |received_message|

  begin
    data = JSON.parse(received_message.data, symbolize_names: true)
    attrs = received_message.attributes
    puts "Message: #{received_message.message_id}"
    puts "Publish time: #{received_message.published_at}"
    puts "Delivery attempt: #{received_message.delivery_attempt}"
    puts "Data: #{data}"
    puts "Attrs: #{attrs}"

    # Process the message
    process_order(data)

    # ACK — confirm successful processing
    received_message.acknowledge!
    puts "ACK: #{received_message.message_id}"

  rescue JSON::ParserError => e
    Rails.logger.error "Invalid JSON: #{e.message}"
    received_message.acknowledge!   # remove invalid messages so they aren't redelivered

  rescue => e
    Rails.logger.error "Error: #{e.message}"
    # NACK — reject, return for redelivery
    received_message.nack!
    # Pub/Sub will retry according to the retry policy (with backoff)
  end
end

# Start the subscriber
subscriber.start

# Graceful shutdown
trap("SIGTERM") do
  subscriber.stop
  subscriber.wait!
  puts "Subscriber stopped"
end

# Block the main thread
subscriber.wait!

sub = pubsub.subscription("orders-analytics")

# Pull a number of messages synchronously
loop do
  received_messages = sub.pull(
    max_messages:      10,     # max messages per pull
    return_immediately: false  # wait if there are no messages
  )

  received_messages.each do |message|
    begin
      data = JSON.parse(message.data, symbolize_names: true)
      process_analytics(data)

      # ACK via the subscription
      sub.acknowledge(message)

    rescue => e
      puts "Error: #{e.message}"
      sub.nack(message)
    end
  end

  sleep 1 if received_messages.empty?
end

# Three options after receiving a message:

# 1. ACK — processed successfully
received_message.acknowledge!
# or: sub.acknowledge(received_message)

# 2. NACK — failed, request redelivery
received_message.nack!
# Pub/Sub will retry according to the backoff policy

# 3. Modify Ack Deadline — extend the time before a retry
# Useful if processing takes longer than the default ack deadline
received_message.modify_ack_deadline!(120)   # add another 120 seconds

# Extend periodically in a separate thread
Thread.new do
  loop do
    sleep 50   # extend before the 60 second deadline expires
    received_message.modify_ack_deadline!(60)
  end
end

# A processing task that takes a long time
heavy_process(JSON.parse(received_message.data))
received_message.acknowledge!

# Set up a Dead Letter Topic
dlq_topic = pubsub.create_topic("orders-dlq")

# Create a DLQ subscription to monitor failed messages
dlq_sub = dlq_topic.subscribe("orders-dlq-monitor")

# Main subscription with a DLQ
sub = topic.subscribe(
  "orders-with-dlq",
  dead_letter_topic:     dlq_topic,
  max_delivery_attempts: 5   # after 5 attempts → DLQ
)

# Monitor the DLQ
dlq_subscriber = dlq_sub.listen do |message|
  attempts = message.delivery_attempt
  puts "DLQ: #{message.message_id} after #{attempts} attempts"

  data = JSON.parse(message.data) rescue message.data

  # Log and save for analysis
  FailedOrder.create!(
    message_id:       message.message_id,
    payload:          message.data,
    delivery_attempt: attempts,
    attributes:       message.attributes,
    failed_at:        Time.now
  )

  # Notify the team
  Monitoring.alert("Message in DLQ", message_id: message.message_id, data: data)

  message.acknowledge!
end

dlq_subscriber.start

sub = pubsub.subscription("orders-analytics")

# Create a snapshot — save the current position
snapshot = sub.create_snapshot("before-migration-#{Time.now.strftime('%Y%m%d%H%M%S')}")
puts "Snapshot: #{snapshot.name}"
puts "Valid until: #{snapshot.expiration_time}"

# Run a migration or experiment
# ...

# If there are problems, roll back to the snapshot
sub.seek(snapshot)   # rewind to the position when the snapshot was created
puts "Seek to snapshot successful"

# Or seek to a specific time (replay messages from 2 hours ago)
sub.seek(2.hours.ago)

# List all snapshots
pubsub.snapshots.each { |s| puts "#{s.name}: #{s.expiration_time}" }

# Delete a snapshot
snapshot.delete

# Main topic receiving all events
topic = pubsub.topic("all_events")

# Subscription that only receives premium orders
premium_sub = topic.subscribe(
  "premium-order-events",
  filter: 'attributes.customer_type = "premium" AND attributes.event_type = "new_order"'
)

# Subscription that only receives payments above 1 million
large_sub = topic.subscribe(
  "large-payment-events",
  filter: 'attributes.event_type = "payment" AND attributes.amount > "1000000"'
)

# Subscription for all errors
error_sub = topic.subscribe(
  "all-error-events",
  filter: 'attributes.severity = "ERROR" OR attributes.severity = "CRITICAL"'
)

# Publishers just publish to one topic with attributes
topic.publish(
  JSON.generate({ order_id: 999, total: 5_000_000 }),
  "event_type"     => "new_order",
  "customer_type"  => "premium",
  "amount"         => "5000000"
)
# Only premium_sub and large_sub will receive this message

# Create a push subscription sending to a Cloud Run endpoint
sub = topic.subscribe(
  "orders-push",
  endpoint:              "https://api.example.com/pubsub/orders",
  authentication: {
    service_account_email: "[email protected]",
    audience:              "https://api.example.com"
  }
)

# On the Rails server side receiving pushes
# app/controllers/pubsub_controller.rb
class PubsubController < ApplicationController
  skip_before_action :verify_authenticity_token

  def orders
    # Pub/Sub sends base64-encoded data in a JSON envelope
    envelope = JSON.parse(request.body.read)
    payload  = Base64.decode64(envelope["message"]["data"])
    data     = JSON.parse(payload, symbolize_names: true)
    attrs    = envelope["message"]["attributes"] || {}

    Rails.logger.info "Pub/Sub message: #{envelope['message']['messageId']}"

    process_order(data)

    # HTTP 200 = ACK. Non-2xx = NACK (Pub/Sub will retry)
    head :ok

  rescue => e
    Rails.logger.error "Error: #{e.message}"
    head :internal_server_error   # trigger a retry
  end
end

# config/initializers/pubsub.rb
module Pubsub
  def self.client
    @client ||= Google::Cloud::PubSub.new(
      project_id: ENV.fetch("GOOGLE_CLOUD_PROJECT"),
      credentials: ENV["GOOGLE_APPLICATION_CREDENTIALS"]
    )
  end

  def self.topic(name)
    @topics ||= {}
    @topics[name] ||= client.topic(name) || client.create_topic(name)
  end

  def self.publish(topic_name, data, **attributes)
    topic(topic_name).publish(
      JSON.generate(data),
      **attributes.transform_keys(&:to_s),
      "environment"   => Rails.env,
      "schema_version" => "v1",
      "timestamp"      => Time.now.iso8601
    )
  end
end

# Use from anywhere in Rails
class OrderService
  def self.create(user, params)
    order = Order.create!(user: user, **params)

    Pubsub.publish(
      "orders",
      { order_id: order.id, total: order.total, status: order.status },
      event_type:      "order_created",
      customer_type:   user.tier
    )

    order
  end
end

# app/workers/pubsub_worker.rb — worker for a pull subscription
class PubsubWorker
  SUBSCRIPTION_NAME = ENV.fetch("PUBSUB_SUBSCRIPTION", "orders-worker")

  def self.run
    sub = Pubsub.client.subscription(SUBSCRIPTION_NAME)

    subscriber = sub.listen(streams: 4, threads: { callback: 8 }) do |msg|
      begin
        data = JSON.parse(msg.data, symbolize_names: true)

        case data[:event]
        when "order_created"
          ProcessOrder.call(data)
        when "payment_success"
          ProcessPayment.call(data)
        else
          Rails.logger.warn "Unknown event: #{data[:event]}"
        end

        msg.acknowledge!

      rescue => e
        Rails.logger.error "PubsubWorker error: #{e.message}\n#{e.backtrace.first(5).join("\n")}"
        msg.nack!
      end
    end

    subscriber.start
    trap("SIGTERM") { subscriber.stop.wait! }
    subscriber.wait!
  end
end

require 'google/cloud/pubsub/v1'

# Pub/Sub Lite uses a different API
lite_client = Google::Cloud::PubSub::V1::AdminService::Client.new

# Create a Lite Topic with provisioned capacity
lite_topic = lite_client.create_topic(
  parent: "projects/my-project/locations/us-central1-a",
  topic_id: "transactions-lite",
  topic: {
    partition_config: {
      count:    8,                    # 8 partitions
      capacity: {
        publish_mib_per_sec:   16,    # 16 MB/s publish capacity
        subscribe_mib_per_sec: 32     # 32 MB/s subscribe capacity
      }
    },
    retention_config: {
      per_partition_bytes: 30.gigabytes,
      period: { seconds: 604_800 }   # 7 days
    }
  }
)

# Important Pub/Sub metrics to monitor:
# subscription/num_undelivered_messages  → number of unprocessed messages
# subscription/oldest_unacked_message_age → age of the oldest unacked message
# topic/send_message_operation_count      → publisher throughput
# subscription/pull_message_operation_count → subscriber throughput

# Create an alert via the gcloud CLI:
# gcloud alpha monitoring policies create \
#   --notification-channels=... \
#   --condition-filter='resource.type="pubsub_subscription"
#     AND metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages"
#     AND metric.labels.subscription_id="orders-notifications"' \
#   --condition-threshold-value=1000 \
#   --condition-threshold-comparison=COMPARISON_GT \
#   --display-name="Pub/Sub orders-notifications high lag"