ballerinax/gcloud.pubsub Ballerina library

Overview

Google Cloud Pub/Sub is an asynchronous messaging service that decouples services that produce events from services that process events. It provides highly reliable and scalable event distribution for modern cloud applications. The GCloud PubSub connector enables seamless integration with Google Cloud Pub/Sub, allowing you to build reactive and event-driven systems.

Key Features

• Highly reliable and scalable asynchronous messaging
• Support for both push and pull message delivery
• Seamless integration with Google Cloud ecosystem
• Secure communication with OAuth2 and IAM
• Simplified management of topics and subscriptions
• GraalVM compatible for native image builds

Setup guide

To use the Ballerina Google Cloud Pub/Sub connector, you need to have a Google Cloud account and a project with the Pub/Sub API enabled.

Create a Google Cloud project

1. Go to the Google Cloud Console.
2. Click on the project dropdown and select New Project.
3. Enter a project name and click Create.
4. Note your project ID, as you'll need it for configuration.

Enable the Pub/Sub API

1. In the Google Cloud Console, navigate to APIs & Services > Library.
2. Search for "Cloud Pub/Sub API".
3. Click on it and then click Enable.

Create a service account and download credentials

1. Navigate to IAM & Admin > Service Accounts in the Google Cloud Console.
2. Click Create Service Account.
3. Enter a service account name and description, then click Create and Continue.
4. Grant the service account the Pub/Sub Publisher and Pub/Sub Subscriber roles.
5. Click Done.
6. Click on the created service account, go to the Keys tab.
7. Click Add Key > Create new key.
8. Select JSON and click Create.
9. Save the downloaded JSON file securely - you'll need the path to this file for authentication.

Create a topic

1. In the Google Cloud Console, navigate to Pub/Sub > Topics.
2. Click Create Topic.
3. Enter the topic ID - my-topic.
4. Click Create.

Create a subscription

1. In the Google Cloud Console, navigate to Pub/Sub > Subscriptions.
2. Click Create Subscription.
3. Enter the subscription ID - my-subscription.
4. Select the topic you created earlier (my-topic) from the dropdown.
5. Choose the delivery type (Pull is recommended for the Ballerina listener).
6. Click Create.

Quickstart

To use the Google Cloud Pub/Sub connector in your Ballerina application, modify the .bal file as follows:

Step 1: Import the connector

Import the ballerinax/gcloud.pubsub module into your Ballerina project.

import ballerinax/gcloud.pubsub;

Step 2: Instantiate a new publisher

Create a pubsub:Publisher instance with your Google Cloud Pub/Sub configuration.

configurable string projectId = ?; // GCP Project ID
configurable string topicName = ?; // Pub/Sub Topic Name
configurable string credentialsPath = ?; // Path to Service Account JSON file

pubsub:Publisher publisher = check new (
    topicName,
    projectId = projectId,
    credentials = {
        credentialsPath: credentialsPath
    }
);

Step 3: Publish messages

Now, utilize the available publisher operations to publish messages.

Publish a simple message

string messageId = check publisher->publish({
    data: "Hello, Google Pub/Sub!".toBytes()
});

Publish a message with attributes

string messageId = check publisher->publish({
    data: "Hello, Google Pub/Sub!".toBytes(),
    attributes: {
        "source": "ballerina-app",
        "version": "1.0"
    }
});

Publish a message with ordering key

check publisher->publish({
    data: "Message 1".toBytes(),
    orderingKey: "customer-123"
});

Publish multiple messages in a batch

string[] messageIds = check publisher->publishBatch([
    {data: "Message 1".toBytes()},
    {data: "Message 2".toBytes()},
    {data: "Message 3".toBytes()}
]);

Step 4: Clean up resources

When done, close the publisher to release resources.

check publisher->close();

Step 5: Set up a listener

Create a pubsub:Listener instance to consume messages from a subscription.

configurable string subscriptionName = ?;

listener pubsub:Listener pubsubListener = check new (
    subscriptionName,
    projectId = projectId,
    credentials = {
        credentialsPath: credentialsPath
    }
);

Step 6: Implement a service to process messages

Attach a service to the listener to process incoming messages.

import ballerina/io;

service on pubsubListener {
    remote function onMessage(pubsub:PubSubMessage message, pubsub:Caller caller) returns error? {
        // Process the message
        io:println("Received message: ", check string:fromBytes(message.data));

        // Print attributes if present
        if message.attributes is map<string> {
            io:println("Attributes: ", message.attributes);
        }

        // Acknowledge the message
        check caller->ack();
    }
}

Handle errors and negative acknowledgements

service on pubsubListener {
    remote function onMessage(pubsub:PubSubMessage message, pubsub:Caller caller) returns error? {
        // Process the message
        error? result = processMessage(message);

        if result is error {
            // If processing fails, nack the message so it can be redelivered
            io:println("Error processing message: ", result.message());
            check caller->nack();
        } else {
            // If processing succeeds, acknowledge the message
            check caller->ack();
        }
    }
}

function processMessage(pubsub:PubSubMessage message) returns error? {
    // Your message processing logic here
    io:println("Processing message: ", check string:fromBytes(<byte[]>message.data));
}

Step 7: Run the Ballerina application

bal run