Receive messages from a pull subscription

namespacepubsub=::google::cloud::pubsub;
autosample=[](pubsub::Subscribersubscriber){
returnsubscriber.Subscribe(
[&](pubsub::Messageconst&m,pubsub::AckHandlerh){
std::cout << "Received message " << m << "\n";
std::move(h).ack();
PleaseIgnoreThisSimplifiesTestingTheSamples();
});
};

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);
}

/**
 * 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
import{PubSub,Message}from'@google-cloud/pubsub';
// Creates a client; cache this for further use
constpubSubClient=newPubSub();
functionlistenForMessages(subscriptionNameOrId:string,timeout:number){
// 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: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);
}

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.

# 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!

namespacepubsub=::google::cloud::pubsub;
autosample=[](pubsub::Subscribersubscriber){
returnsubscriber.Subscribe(
[&](pubsub::Messageconst&m,pubsub::AckHandlerh){
std::cout << "Received message with attributes:\n";
for(auto&kv:m.attributes()){
std::cout << " " << kv.first << ": " << kv.second << "\n";
}
std::move(h).ack();
PleaseIgnoreThisSimplifiesTestingTheSamples();
});
};

usingGoogle.Cloud.PubSub.V1 ;
usingSystem;
usingSystem.Collections.Generic;
usingSystem.Threading;
usingSystem.Threading.Tasks;
publicclassPullMessagesWithCustomAttributesAsyncSample
{
publicasyncTask<List<PubsubMessage>>PullMessagesWithCustomAttributesAsync(stringprojectId,stringsubscriptionId,boolacknowledge)
{
SubscriptionName subscriptionName=SubscriptionName .FromProjectSubscription (projectId,subscriptionId);
SubscriberClient subscriber=awaitSubscriberClient .CreateAsync (subscriptionName);
varmessages=newList<PubsubMessage>();
TaskstartTask=subscriber.StartAsync ((PubsubMessagemessage,CancellationTokencancel)=>
{
messages.Add(message);
stringtext=message.Data .ToStringUtf8();
Console.WriteLine($"Message {message.MessageId}: {text}");
if(message.Attributes!=null)
{
foreach(varattributeinmessage.Attributes)
{
Console.WriteLine($"{attribute.Key} = {attribute.Value}");
}
}
returnTask.FromResult(acknowledge?SubscriberClient .Reply .Ack :SubscriberClient .Reply .Nack );
});
// Run for 7 seconds.
awaitTask.Delay(7000);
awaitsubscriber.StopAsync (CancellationToken.None);
// Lets make sure that the start task finished successfully after the call to stop.
awaitstartTask;
returnmessages;
}
}

import(
"context"
"fmt"
"io"
"time"
"cloud.google.com/go/pubsub/v2"
)
funcpullMsgsCustomAttributes(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()
// Receive blocks until the context is cancelled or an error occurs.
err=sub.Receive(ctx,func(_context.Context,msg*pubsub.Message){
fmt.Fprintf(w,"Got message :%q\n",string(msg.Data))
fmt.Fprintln(w,"Attributes:")
forkey,value:=rangemsg.Attributes{
fmt.Fprintf(w,"%s = %s\n",key,value)
}
msg.Ack()
})
iferr!=nil{
returnfmt.Errorf("sub.Receive: %w",err)
}
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 SubscribeWithCustomAttributesExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
StringsubscriptionId="your-subscription-id";
subscribeWithCustomAttributesExample(projectId,subscriptionId);
}
publicstaticvoidsubscribeWithCustomAttributesExample(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());
// Print message attributes.
message
.getAttributesMap()
.forEach((key,value)->System.out.println(key+" = "+value));
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 ();
asyncfunctionlistenWithCustomAttributes(subscriptionNameOrId,timeout){
// References an existing subscription, e.g. "my-subscription"
constsubscription=pubSubClient.subscription(subscriptionNameOrId);
// Create an event handler to handle messages
constmessageHandler=message=>{
console.log(
`Received message: id ${message.id}, data ${
message.data
}, attributes: ${JSON.stringify(message.attributes)}`,
);
// "Ack" (acknowledge receipt of) the message
message.ack ();
};
// Wait a while for the subscription to run. (Part of the sample only.)
subscription .on ('message',messageHandler);
setTimeout(()=>{
subscription.removeListener('message',messageHandler);
},timeout*1000);
}

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 ()
subscription_path = subscriber.subscription_path(project_id, subscription_id)
defcallback(message: pubsub_v1.subscriber.message.Message ) -> None:
 print(f"Received {message.data !r}.")
 if message.attributes :
 print("Attributes:")
 for key in message.attributes :
 value = message.attributes .get(key)
 print(f"{key}: {value}")
 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.

# 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}"
unlessreceived_message.attributes.empty?
puts"Attributes:"
received_message.attributes.eachdo|key,value|
puts"#{key}: #{value}"
end
end
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!

namespacepubsub=::google::cloud::pubsub;
using::google::cloud::future;
autosample=[](pubsub::Subscribersubscriber){
returnsubscriber
.Subscribe([&](pubsub::Messageconst&m,pubsub::AckHandlerh){
std::cout << "Received message " << m << "\n";
std::move(h).ack();
PleaseIgnoreThisSimplifiesTestingTheSamples();
})
// Setup an error handler for the subscription session
.then([](future<google::cloud::Status>f){
std::cout << "Subscription session result: " << f.get() << "\n";
});
};

import(
"context"
"fmt"
"io"
"cloud.google.com/go/pubsub/v2"
)
funcpullMsgsError(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()
// If the service returns a non-retryable error, Receive returns that error after
// all of the outstanding calls to the handler have returned.
// 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)
err=sub.Receive(ctx,func(ctxcontext.Context,msg*pubsub.Message){
fmt.Fprintf(w,"Got message: %q\n",string(msg.Data))
msg.Ack()
})
iferr!=nil{
returnfmt.Errorf("Receive: %w",err)
}
returnnil
}

importcom.google.api.gax.core.ExecutorProvider ;
importcom.google.api.gax.core.InstantiatingExecutorProvider ;
importcom.google.cloud.pubsub.v1.AckReplyConsumer ;
importcom.google.cloud.pubsub.v1.MessageReceiver ;
importcom.google.cloud.pubsub.v1.Subscriber ;
importcom.google.common.util.concurrent.MoreExecutors;
importcom.google.pubsub.v1.ProjectSubscriptionName ;
importcom.google.pubsub.v1.PubsubMessage ;
importjava.util.concurrent.TimeUnit;
importjava.util.concurrent.TimeoutException;
publicclass SubscribeWithErrorListenerExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
StringsubscriptionId="your-subscription-id";
subscribeWithErrorListenerExample(projectId,subscriptionId);
}
publicstaticvoidsubscribeWithErrorListenerExample(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{
// Provides an executor service for processing messages.
ExecutorProvider executorProvider=
InstantiatingExecutorProvider .newBuilder().setExecutorThreadCount (4).build();
subscriber=
Subscriber .newBuilder(subscriptionName,receiver)
.setExecutorProvider(executorProvider)
.build();
// Listen for unrecoverable failures.
subscriber.addListener(
newSubscriber .Listener(){
publicvoidfailed(Subscriber .Statefrom,Throwablefailure){
System.out.println("Unrecoverable subscriber failure:"+failure.getStackTrace());
}
},
MoreExecutors.directExecutor());
// 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 = 10;
// Imports the Google Cloud client library
const{PubSub}=require('@google-cloud/pubsub');
// Creates a client; cache this for further use
constpubSubClient=newPubSub ();
functionlistenForErrors(subscriptionNameOrId,timeout){
// References an existing subscription
constsubscription=pubSubClient.subscription(subscriptionNameOrId);
// Create an event handler to handle messages
constmessageHandler=message=>{
// Do something with the message
console.log(`Message: ${message}`);
// "Ack" (acknowledge receipt of) the message
message.ack ();
};
// Create an event handler to handle errors
consterrorHandler=error=>{
// Do something with the error
console.error(`ERROR: ${error}`);
throwerror;
};
// Listen for new messages/errors until timeout is hit
subscription .on ('message',messageHandler);
subscription .on ('error',errorHandler);
// Wait a while for the subscription to run. (Part of the sample only.)
setTimeout(()=>{
subscription.removeListener('message',messageHandler);
subscription.removeListener('error',errorHandler);
},timeout*1000);
}

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 ()
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:
 # When `timeout` is not set, result() will block indefinitely,
 # unless an exception is encountered first.
 try:
 streaming_pull_future.result(timeout=timeout)
 except Exception as e:
 print(
 f"Listening for messages on {subscription_path} threw an exception: {e}."
 )
 streaming_pull_future.cancel() # Trigger the shutdown.
 streaming_pull_future.result() # Block until the shutdown is complete.

# 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
# Propagate exception from child threads to the main thread as soon as it is
# raised. Exceptions happened in the callback thread are collected in the
# callback thread pool and do not propagate to the main thread
Thread.abort_on_exception=true
begin
listener.start
# Let the main thread sleep for 60 seconds so the thread for listening
# messages does not quit
sleep60
listener.stop.wait!
rescueStandardError=>e
puts"Exception #{e.inspect}: #{e.message}"
raise"Stopped listening for messages."
end