Publish and receive messages in Pub/Sub by using a client library

<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>libraries-bom</artifactId>
<version>26.76.0</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>google-cloud-pubsub</artifactId>
</dependency>
</dependencies>

implementationplatform('com.google.cloud:libraries-bom:26.80.0')
implementation'com.google.cloud:google-cloud-pubsub'

libraryDependencies+="com.google.cloud"%"google-cloud-pubsub"%"1.150.2"

fromgoogle.cloudimport pubsub_v1
# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
publisher = pubsub_v1.PublisherClient ()
# The `topic_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/topics/{topic_id}`
topic_path = publisher.topic_path(project_id, topic_id)
for n in range(1, 10):
 data_str = f"Message number {n}"
 # Data must be a bytestring
 data = data_str.encode("utf-8")
 # When you publish a message, the client returns a future.
 future = publisher.publish (topic_path, data)
 print(future.result())
print(f"Published messages to {topic_path}.")

#include"google/cloud/pubsub/publisher.h"
#include <iostream>
intmain(intargc,char*argv[])try{
if(argc!=3){
std::cerr << "Usage: " << argv[0] << " <project-id> <topic-id>\n";
return1;
}
std::stringconstproject_id=argv[1];
std::stringconsttopic_id=argv[2];
// Create a namespace alias to make the code easier to read.
namespacepubsub=::google::cloud::pubsub;
autopublisher=pubsub::Publisher(
pubsub::MakePublisherConnection(pubsub::Topic(project_id,topic_id)));
autoid=
publisher
.Publish(pubsub::MessageBuilder{}.SetData("Hello World!").Build())
.get();
if(!id)throwstd::move(id).status();
std::cout << "Hello World published with id=" << *id << "\n";
return0;
}catch(google::cloud::Statusconst&status){
std::cerr << "google::cloud::Status thrown: " << status << "\n";
return1;
}

usingGoogle.Cloud.PubSub.V1 ;
usingSystem;
usingSystem.Collections.Generic;
usingSystem.Linq;
usingSystem.Threading;
usingSystem.Threading.Tasks;
publicclassPublishMessagesAsyncSample
{
publicasyncTask<int>PublishMessagesAsync(stringprojectId,stringtopicId,IEnumerable<string>messageTexts)
{
TopicName topicName=TopicName .FromProjectTopic (projectId,topicId);
PublisherClient publisher=awaitPublisherClient .CreateAsync (topicName);
intpublishedMessageCount=0;
varpublishTasks=messageTexts.Select(asynctext=>
{
try
{
stringmessage=awaitpublisher.PublishAsync(text);
Console.WriteLine($"Published message {message}");
Interlocked.Increment(refpublishedMessageCount);
}
catch(Exceptionexception)
{
Console.WriteLine($"An error occurred when publishing message {text}: {exception.Message}");
}
});
awaitTask.WhenAll(publishTasks);
// PublisherClient instance should be shutdown after use.
// The TimeSpan specifies for how long to attempt to publish locally queued messages.
awaitpublisher.ShutdownAsync (TimeSpan.FromSeconds(15));
returnpublishedMessageCount;
}
}

import(
"context"
"fmt"
"io"
"cloud.google.com/go/pubsub/v2"
)
funcpublish(wio.Writer,projectID,topicID,msgstring)error{
// projectID := "my-project-id"
// topicID := "my-topic"
// msg := "Hello World"
ctx:=context.Background()
client,err:=pubsub.NewClient(ctx,projectID)
iferr!=nil{
returnfmt.Errorf("pubsub: NewClient: %w",err)
}
deferclient.Close()
// client.Publisher can be passed a topic ID (e.g. "my-topic") or
// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
// If a topic ID is provided, the project ID from the client is used.
// Reuse this publisher for all publish calls to send messages in batches.
publisher:=client.Publisher(topicID)
result:=publisher.Publish(ctx,&pubsub.Message{
Data:[]byte(msg),
})
// Block until the result is returned and a server-generated
// ID is returned for the published message.
id,err:=result.Get(ctx)
iferr!=nil{
returnfmt.Errorf("pubsub: result.Get: %w",err)
}
fmt.Fprintf(w,"Published a message; msg ID: %v\n",id)
returnnil
}

importcom.google.api.core.ApiFuture ;
importcom.google.cloud.pubsub.v1.Publisher ;
importcom.google.protobuf.ByteString ;
importcom.google.pubsub.v1.PubsubMessage ;
importcom.google.pubsub.v1.TopicName ;
importjava.io.IOException;
importjava.util.concurrent.ExecutionException;
importjava.util.concurrent.TimeUnit;
publicclass PublisherExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
StringtopicId="your-topic-id";
publisherExample(projectId,topicId);
}
publicstaticvoidpublisherExample(StringprojectId,StringtopicId)
throwsIOException,ExecutionException,InterruptedException{
TopicName topicName=TopicName .of(projectId,topicId);
Publisher publisher=null;
try{
// Create a publisher instance with default settings bound to the topic
publisher=Publisher .newBuilder(topicName).build();
Stringmessage="Hello World!";
ByteString data=ByteString .copyFromUtf8 (message);
PubsubMessage pubsubMessage=PubsubMessage .newBuilder().setData (data).build();
// Once published, returns a server-assigned message id (unique within the topic)
ApiFuture<String>messageIdFuture=publisher.publish (pubsubMessage);
StringmessageId=messageIdFuture.get();
System.out.println("Published message ID: "+messageId);
}finally{
if(publisher!=null){
// When finished with the publisher, shutdown to free up resources.
publisher.shutdown ();
publisher.awaitTermination (1,TimeUnit.MINUTES);
}
}
}
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const data = JSON.stringify({foo: 'bar'});
// Imports the Google Cloud client library
const{PubSub}=require('@google-cloud/pubsub');
// Creates a client; cache this for further use
constpubSubClient=newPubSub ();
asyncfunctionpublishMessage(topicNameOrId,data){
// Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
constdataBuffer=Buffer.from (data);
// Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId);
try{
constmessageId=awaittopic.publishMessage ({data:dataBuffer});
console.log(`Message ${messageId} published.`);
}catch(error){
console.error(`Received error while publishing: ${error.message}`);
process.exitCode=1;
}
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const data = JSON.stringify({foo: 'bar'});
// Imports the Google Cloud client library
import{PubSub}from'@google-cloud/pubsub';
// Creates a client; cache this for further use
constpubSubClient=newPubSub();
asyncfunctionpublishMessage(topicNameOrId:string,data:string){
// Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
constdataBuffer=Buffer.from(data);
// Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId);
try{
constmessageId=awaittopic.publishMessage({data:dataBuffer});
console.log(`Message ${messageId} published.`);
}catch(error){
console.error(
`Received error while publishing: ${(errorasError).message}`,
);
process.exitCode=1;
}
}

use Google\Cloud\PubSub\MessageBuilder;
use Google\Cloud\PubSub\PubSubClient;
/**
 * Publishes a message for a Pub/Sub topic.
 *
 * @param string $projectId The Google project ID.
 * @param string $topicName The Pub/Sub topic name.
 * @param string $message The message to publish.
 */
function publish_message($projectId, $topicName, $message)
{
 $pubsub = new PubSubClient([
 'projectId' => $projectId,
 ]);
 $topic = $pubsub->topic($topicName);
 $topic->publish((new MessageBuilder)->setData($message)->build());
 print('Message published' . PHP_EOL);
}

# topic_id = "your-topic-id"
pubsub=Google::Cloud::PubSub .new
publisher=pubsub.publisher topic_id
publisher.publish"This is a test message."
puts"Message published."

fromconcurrent.futuresimport TimeoutError
fromgoogle.cloudimport pubsub_v1
# TODO(developer)
# project_id = "your-project-id"
# subscription_id = "your-subscription-id"
# Number of seconds the subscriber should listen for messages
# timeout = 5.0
subscriber = pubsub_v1.SubscriberClient ()
# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)
defcallback(message: pubsub_v1.subscriber.message.Message ) -> None:
 print(f"Received {message}.")
 message.ack ()
streaming_pull_future = subscriber.subscribe (subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}..\n")
# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
 try:
 # When `timeout` is not set, result() will block indefinitely,
 # unless an exception is encountered first.
 streaming_pull_future.result(timeout=timeout)
 except TimeoutError:
 streaming_pull_future.cancel() # Trigger the shutdown.
 streaming_pull_future.result() # Block until the shutdown is complete.

#include"google/cloud/pubsub/message.h"
#include"google/cloud/pubsub/subscriber.h"
#include <iostream>
intmain(intargc,char*argv[])try{
if(argc!=3){
std::cerr << "Usage: " << argv[0] << " <project-id> <subscription-id>\n";
return1;
}
std::stringconstproject_id=argv[1];
std::stringconstsubscription_id=argv[2];
autoconstexprkWaitTimeout=std::chrono::seconds(30);
// Create a namespace alias to make the code easier to read.
namespacepubsub=::google::cloud::pubsub;
autosubscriber=pubsub::Subscriber(pubsub::MakeSubscriberConnection(
pubsub::Subscription(project_id,subscription_id)));
autosession=
subscriber.Subscribe([&](pubsub::Messageconst&m,pubsub::AckHandlerh){
std::cout << "Received message " << m << "\n";
std::move(h).ack();
});
std::cout << "Waiting for messages on "+subscription_id+"...\n";
// Blocks until the timeout is reached.
autoresult=session.wait_for(kWaitTimeout);
if(result==std::future_status::timeout){
std::cout << "timeout reached, ending session\n";
session.cancel();
}
return0;
}catch(google::cloud::Statusconst&status){
std::cerr << "google::cloud::Status thrown: " << status << "\n";
return1;
}

usingGoogle.Cloud.PubSub.V1 ;
usingSystem;
usingSystem.Threading;
usingSystem.Threading.Tasks;
publicclassPullMessagesAsyncSample
{
publicasyncTask<int>PullMessagesAsync(stringprojectId,stringsubscriptionId,boolacknowledge)
{
SubscriptionName subscriptionName=SubscriptionName .FromProjectSubscription (projectId,subscriptionId);
SubscriberClient subscriber=awaitSubscriberClient .CreateAsync (subscriptionName);
// SubscriberClient runs your message handle function on multiple
// threads to maximize throughput.
intmessageCount=0;
TaskstartTask=subscriber.StartAsync ((PubsubMessage message,CancellationTokencancel)=>
{
stringtext=message.Data.ToStringUtf8();
Console.WriteLine($"Message {message.MessageId}: {text}");
Interlocked.Increment(refmessageCount);
returnTask.FromResult(acknowledge?SubscriberClient .Reply .Ack :SubscriberClient .Reply .Nack );
});
// Run for 5 seconds.
awaitTask.Delay(5000);
awaitsubscriber.StopAsync (CancellationToken.None);
// Lets make sure that the start task finished successfully after the call to stop.
awaitstartTask;
returnmessageCount;
}
}

import(
"context"
"fmt"
"io"
"sync/atomic"
"time"
"cloud.google.com/go/pubsub/v2"
)
funcpullMsgs(wio.Writer,projectID,subIDstring)error{
// projectID := "my-project-id"
// subID := "my-sub"
ctx:=context.Background()
client,err:=pubsub.NewClient(ctx,projectID)
iferr!=nil{
returnfmt.Errorf("pubsub.NewClient: %w",err)
}
deferclient.Close()
// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
// If a subscription ID is provided, the project ID from the client is used.
sub:=client.Subscriber(subID)
// Receive messages for 10 seconds, which simplifies testing.
// Comment this out in production, since `Receive` should
// be used as a long running operation.
ctx,cancel:=context.WithTimeout(ctx,10*time.Second)
defercancel()
varreceivedint32
err=sub.Receive(ctx,func(_context.Context,msg*pubsub.Message){
fmt.Fprintf(w,"Got message: %q\n",string(msg.Data))
atomic.AddInt32(&received,1)
msg.Ack()
})
iferr!=nil{
returnfmt.Errorf("sub.Receive: %w",err)
}
fmt.Fprintf(w,"Received %d messages\n",received)
returnnil
}

importcom.google.cloud.pubsub.v1.AckReplyConsumer ;
importcom.google.cloud.pubsub.v1.MessageReceiver ;
importcom.google.cloud.pubsub.v1.Subscriber ;
importcom.google.pubsub.v1.ProjectSubscriptionName ;
importcom.google.pubsub.v1.PubsubMessage ;
importjava.util.concurrent.TimeUnit;
importjava.util.concurrent.TimeoutException;
publicclass SubscribeAsyncExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
StringsubscriptionId="your-subscription-id";
subscribeAsyncExample(projectId,subscriptionId);
}
publicstaticvoidsubscribeAsyncExample(StringprojectId,StringsubscriptionId){
ProjectSubscriptionName subscriptionName=
ProjectSubscriptionName .of(projectId,subscriptionId);
// Instantiate an asynchronous message receiver.
MessageReceiver receiver=
(PubsubMessagemessage,AckReplyConsumerconsumer)->{
// Handle incoming message, then ack the received message.
System.out.println("Id: "+message.getMessageId());
System.out.println("Data: "+message.getData().toStringUtf8());
consumer.ack();
};
Subscriber subscriber=null;
try{
subscriber=Subscriber .newBuilder(subscriptionName,receiver).build();
// Start the subscriber.
subscriber.startAsync ().awaitRunning();
System.out.printf("Listening for messages on %s:\n",subscriptionName.toString ());
// Allow the subscriber to run for 30s unless an unrecoverable error occurs.
subscriber.awaitTerminated(30,TimeUnit.SECONDS);
}catch(TimeoutExceptiontimeoutException){
// Shut down the subscriber after 30s. Stop receiving messages.
subscriber.stopAsync();
}
}
}

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const timeout = 60;
// Imports the Google Cloud client library
const{PubSub}=require('@google-cloud/pubsub');
// Creates a client; cache this for further use
constpubSubClient=newPubSub ();
functionlistenForMessages(subscriptionNameOrId,timeout){
// References an existing subscription; if you are unsure if the
// subscription will exist, try the optimisticSubscribe sample.
constsubscription=pubSubClient.subscription(subscriptionNameOrId);
// Create an event handler to handle messages
letmessageCount=0;
constmessageHandler=message=>{
console.log(`Received message ${message.id}:`);
console.log(`\tData: ${message.data}`);
console.log(`\tAttributes: ${message.attributes}`);
messageCount+=1;
// "Ack" (acknowledge receipt of) the message
message.ack ();
};
// Listen for new messages until timeout is hit
subscription .on ('message',messageHandler);
// Wait a while for the subscription to run. (Part of the sample only.)
setTimeout(()=>{
subscription.removeListener('message',messageHandler);
console.log(`${messageCount} message(s) received.`);
},timeout*1000);
}

use Google\Cloud\PubSub\PubSubClient;
/**
 * Pulls all Pub/Sub messages for a subscription.
 *
 * @param string $projectId The Google project ID.
 * @param string $subscriptionName The Pub/Sub subscription name.
 */
function pull_messages($projectId, $subscriptionName)
{
 $pubsub = new PubSubClient([
 'projectId' => $projectId,
 ]);
 $subscription = $pubsub->subscription($subscriptionName);
 foreach ($subscription->pull() as $message) {
 printf('Message: %s' . PHP_EOL, $message->data());
 // Acknowledge the Pub/Sub message has been received, so it will not be pulled multiple times.
 $subscription->acknowledge($message);
 }
}

# subscription_id = "your-subscription-id"
pubsub=Google::Cloud::PubSub .new
subscriber=pubsub.subscriber subscription_id
listener=subscriber.listen do|received_message|
puts"Received message: #{received_message.data}"
received_message.acknowledge!
end
listener.start
# Let the main thread sleep for 60 seconds so the thread for listening
# messages does not quit
sleep60
listener.stop.wait!