Google Pub/Sub component

The google-pubsub component publishes messages to Google Cloud Pub/Sub topics and consumes messages from Pub/Sub subscriptions, with gRPC streaming pull by default.

Scheme: google-pubsub: · Crate: cameleer-cloud · Cargo feature: cloud · Consumer: yes · Producer: yes · Apache Camel: Google Pubsub

When to use it

• Turn webhooks or HTTP calls into Pub/Sub events for downstream services.
• Consume a subscription and write each message to a database, a file or another system.
• Connect edge gateways to Google Cloud: publish readings from the device, consume commands.
• Feed AI agents: publish tasks to a topic and let several workers consume the subscription.

URI format

google-pubsub://projectId:destinationName[?options]

The destination is a topic for the producer and a subscription for the consumer.

Example

Accept JSON events on an HTTP endpoint and publish each one to a topic:

- route:
    id: webhook-to-pubsub
    from:
      uri: "platform-http:/events?httpMethodRestrict=POST"
      steps:
        - to: "google-pubsub://{{env:GCP_PROJECT:my-project}}:{{env:PUBSUB_TOPIC:events}}"
        - setHeader:
            name: CamelHttpResponseCode
            constant: 202
        - setBody:
            simple: "${header.CamelGooglePubsubMessageId}"

Consume a subscription (streaming pull) and write every message to a file named after its message id. Successful exchanges are acknowledged, failed ones are nacked and redelivered:

- route:
    id: pubsub-to-files
    from:
      uri: "google-pubsub://{{env:GCP_PROJECT:my-project}}:{{env:PUBSUB_SUBSCRIPTION:events-sub}}?concurrentConsumers=2"
      steps:
        - log: "message ${header.CamelGooglePubsubMessageId} published at ${header.CamelGooglePubsubPublishTime}"
        - setHeader:
            name: CamelFileName
            simple: "${header.CamelGooglePubsubMessageId}.json"
        - to: "file:{{env:OUT_DIR:out}}"

Rust DSL

ctx.add_routes(|b| {
    b.from("google-pubsub://my-project:events-sub")
        .log("got ${body}")
        .to("google-pubsub://my-project:audit");
})?;

Options

Credentials and endpoint options (serviceAccountKey, GOOGLE_APPLICATION_CREDENTIALS, useMetadataServer, authenticate=false and endpoint for the Pub/Sub emulator, ...) are the same for every Google component: see credentials.

Message headers

A list body is published as one message per element. The consumer's body is the message data as bytes.

Run it with Docker

docker run --rm -p 8080:8080 -v "$PWD:/routes:ro" \
  -v "$HOME/keys/sa.json:/keys/sa.json:ro" -e GOOGLE_APPLICATION_CREDENTIALS=/keys/sa.json \
  -e GCP_PROJECT=my-project -e PUBSUB_TOPIC=events \
  pmsudhi/cameleer run /routes/webhook-to-pubsub.camel.yaml

Common questions

How do I use the Pub/Sub emulator?

Add authenticate=false&endpoint=localhost:8085. The REST calls and streaming pull both go to that address (plaintext); pubsubEndpoint overrides the streaming pull target alone.

How do I set message attributes?

Put a map in the CamelGooglePubsubAttributes header before the producer, for example with a processor or a setHeader whose expression returns a map.

How do I create a subscription with a dead-letter topic?

Send a message to google-pubsub://project:my-sub?operation=createSubscription&topic=orders&deadLetterTopic=orders-dlq&maxDeliveryAttempts=5.

Why does my consumer get each message only once even when it fails?

It does not: with ackMode=AUTO a failed exchange is nacked, and Pub/Sub delivers it again. Use the error handler or a dead-letter policy to stop endless retries.

Related

• Google Pub/Sub Lite, Google BigQuery
• Kafka for self-hosted streaming
• Apache Camel Google Pubsub

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.