Flow control

namespacepubsub=::google::cloud::pubsub;
using::google::cloud::future;
using::google::cloud::Options;
using::google::cloud::StatusOr;
[](std::stringproject_id,std::stringtopic_id){
autotopic=pubsub::Topic(std::move(project_id),std::move(topic_id));
// Configure the publisher to block if either (1) 100 or more messages, or
// (2) messages with 100MiB worth of data have not been acknowledged by the
// service. By default the publisher never blocks, and its capacity is only
// limited by the system's memory.
autopublisher=pubsub::Publisher(pubsub::MakePublisherConnection(
std::move(topic),
Options{}
.set<pubsub::MaxPendingMessagesOption>(100)
.set<pubsub::MaxPendingBytesOption>(100*1024*1024L)
.set<pubsub::FullPublisherActionOption>(
pubsub::FullPublisherAction::kBlocks)));
std::vector<future<void>>ids;
for(charconst*data:{"a","b","c"}){
ids.push_back(
publisher.Publish(pubsub::MessageBuilder().SetData(data).Build())
.then([data](future<StatusOr<std::string>>f){
autos=f.get();
if(!s)return;
std::cout << "Sent '" << data << "' (" << *s << ")\n";
}));
}
publisher.Flush();
// Block until they are actually sent.
for(auto&id:ids)id.get();
}

import(
"context"
"fmt"
"io"
"strconv"
"sync"
"sync/atomic"
"cloud.google.com/go/pubsub/v2"
)
funcpublishWithFlowControlSettings(wio.Writer,projectID,topicIDstring)error{
// projectID := "my-project-id"
// topicID := "my-topic"
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)
publisher.PublishSettings.FlowControlSettings=pubsub.FlowControlSettings{
MaxOutstandingMessages:100,// default 1000
MaxOutstandingBytes:10*1024*1024,// default 0 (unlimited)
LimitExceededBehavior:pubsub.FlowControlBlock,// default Ignore, other options: Block and SignalError
}
varwgsync.WaitGroup
vartotalErrorsuint64
numMsgs:=1000
// Rapidly publishing 1000 messages in a loop may be constrained by flow control.
fori:=0;i < numMsgs;i++{
wg.Add(1)
result:=publisher.Publish(ctx,&pubsub.Message{
Data:[]byte("message #"+strconv.Itoa(i)),
})
gofunc(iint,res*pubsub.PublishResult){
fmt.Fprintf(w,"Publishing message %d\n",i)
deferwg.Done()
// The Get method blocks until a server-generated ID or
// an error is returned for the published message.
_,err:=res.Get(ctx)
iferr!=nil{
// Error handling code can be added here.
fmt.Fprintf(w,"Failed to publish: %v",err)
atomic.AddUint64(&totalErrors,1)
return
}
}(i,result)
}
wg.Wait()
iftotalErrors > 0{
returnfmt.Errorf("%d of %d messages did not publish successfully",totalErrors,numMsgs)
}
returnnil
}

importcom.google.api.core.ApiFuture ;
importcom.google.api.core.ApiFutures ;
importcom.google.api.gax.batching.BatchingSettings ;
importcom.google.api.gax.batching.FlowControlSettings ;
importcom.google.api.gax.batching.FlowController.LimitExceededBehavior ;
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.ArrayList;
importjava.util.List;
importjava.util.concurrent.ExecutionException;
importjava.util.concurrent.TimeUnit;
publicclass PublishWithFlowControlExample{
publicstaticvoidmain(String...args)throwsException{
// TODO(developer): Replace these variables before running the sample.
StringprojectId="your-project-id";
StringtopicId="your-topic-id";
publishWithFlowControlExample(projectId,topicId);
}
publicstaticvoidpublishWithFlowControlExample(StringprojectId,StringtopicId)
throwsIOException,ExecutionException,InterruptedException{
TopicName topicName=TopicName .of(projectId,topicId);
Publisher publisher=null;
List<ApiFuture<String>>messageIdFutures=newArrayList<>();
try{
// Configure how many messages the publisher client can hold in memory
// and what to do when messages exceed the limit.
FlowControlSettings flowControlSettings=
FlowControlSettings .newBuilder()
// Block more messages from being published when the limit is reached. The other
// options are Ignore (or continue publishing) and ThrowException (or error out).
.setLimitExceededBehavior(LimitExceededBehavior .Block)
.setMaxOutstandingRequestBytes(10*1024*1024L)// 10 MiB
.setMaxOutstandingElementCount(100L)// 100 messages
.build();
// By default, messages are not batched.
BatchingSettings batchingSettings=
BatchingSettings .newBuilder().setFlowControlSettings(flowControlSettings).build();
publisher=Publisher .newBuilder(topicName).setBatchingSettings(batchingSettings).build();
// Publish 1000 messages in quick succession may be constrained by publisher flow control.
for(inti=0;i < 1000;i++){
Stringmessage="message "+i;
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);
messageIdFutures.add(messageIdFuture);
}
}finally{
// Wait on any pending publish requests.
List<String>messageIds=ApiFutures .allAsList (messageIdFutures).get();
System.out.println(
"Published "+messageIds.size()+" messages with flow control settings.");
if(publisher!=null){
// When finished with the publisher, shut down to free up resources.
publisher.shutdown ();
publisher.awaitTermination (1,TimeUnit.MINUTES);
}
}
}
}

/**
 * TODO(developer): Uncomment this variable before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// Imports the Google Cloud client library
const{PubSub}=require('@google-cloud/pubsub');
// Creates a client; cache this for further use
constpubSubClient=newPubSub ();
asyncfunctionpublishWithFlowControl(topicNameOrId){
// Create publisher options
constoptions={
flowControlOptions:{
maxOutstandingMessages:50,
maxOutstandingBytes:10*1024*1024,// 10 MB
},
};
// Get a publisher. Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId,options);
// For flow controlled publishing, we'll use a publisher flow controller
// instead of `topic.publish()`.
constflow=topic.flowControlled ();
// Publish messages in a fast loop.
consttestMessage={data:Buffer.from ('test!')};
for(leti=0;i < 1000;i++){
// You can also just `await` on `publish()` unconditionally, but if
// you want to avoid pausing to the event loop on each iteration,
// you can manually check the return value before doing so.
constwait=flow.publish(testMessage);
if(wait){
awaitwait;
}
}
// Wait on any pending publish requests. Note that you can call `all()`
// earlier if you like, and it will return a Promise for all messages
// that have been sent to `flowController.publish()` so far.
constmessageIds=awaitflow.all();
console.log(`Published ${messageIds.length } with flow control settings.`);
}

/**
 * TODO(developer): Uncomment this variable before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// Imports the Google Cloud client library
import{PubSub,PublishOptions}from'@google-cloud/pubsub';
// Creates a client; cache this for further use
constpubSubClient=newPubSub();
asyncfunctionpublishWithFlowControl(topicNameOrId:string){
// Create publisher options
constoptions:PublishOptions={
flowControlOptions:{
maxOutstandingMessages:50,
maxOutstandingBytes:10*1024*1024,// 10 MB
},
};
// Get a publisher. Cache topic objects (publishers) and reuse them.
consttopic=pubSubClient.topic(topicNameOrId,options);
// For flow controlled publishing, we'll use a publisher flow controller
// instead of `topic.publish()`.
constflow=topic.flowControlled();
// Publish messages in a fast loop.
consttestMessage={data:Buffer.from('test!')};
for(leti=0;i < 1000;i++){
// You can also just `await` on `publish()` unconditionally, but if
// you want to avoid pausing to the event loop on each iteration,
// you can manually check the return value before doing so.
constwait=flow.publish(testMessage);
if(wait){
awaitwait;
}
}
// Wait on any pending publish requests. Note that you can call `all()`
// earlier if you like, and it will return a Promise for all messages
// that have been sent to `flowController.publish()` so far.
constmessageIds=awaitflow.all();
console.log(`Published ${messageIds.length} with flow control settings.`);
}

fromconcurrentimport futures
fromgoogle.cloudimport pubsub_v1
fromgoogle.cloud.pubsub_v1.typesimport (
 LimitExceededBehavior ,
 PublisherOptions ,
 PublishFlowControl ,
)
# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# Configure how many messages the publisher client can hold in memory
# and what to do when messages exceed the limit.
flow_control_settings = PublishFlowControl(
 message_limit=100, # 100 messages
 byte_limit=10 * 1024 * 1024, # 10 MiB
 limit_exceeded_behavior=LimitExceededBehavior .BLOCK,
)
publisher = pubsub_v1.PublisherClient (
 publisher_options=PublisherOptions(flow_control=flow_control_settings)
)
topic_path = publisher.topic_path(project_id, topic_id)
publish_futures = []
# Resolve the publish future in a separate thread.
defcallback(publish_future: pubsub_v1.publisher.futures.Future) -> None:
 message_id = publish_future.result ()
 print(message_id)
# Publish 1000 messages in quick succession may be constrained by
# publisher flow control.
for n in range(1, 1000):
 data_str = f"Message number {n}"
 # Data must be a bytestring
 data = data_str.encode("utf-8")
 publish_future = publisher.publish (topic_path, data)
 # Non-blocking. Allow the publisher client to batch messages.
 publish_future.add_done_callback (callback)
 publish_futures.append(publish_future)
futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)
print(f"Published messages with flow control settings to {topic_path}.")

# topic_id = "your-topic-id"
pubsub=Google::Cloud::PubSub .new
publisher=pubsub.publisher topic_id,async:{
# Configure how many messages the publisher client can hold in memory
# and what to do when messages exceed the limit.
flow_control:{
message_limit:100,
byte_limit:10*1024*1024,# 10 MiB
# Block more messages from being published when the limit is reached. The
# other options are :ignore and :error.
limit_exceeded_behavior::block
}
}
# Rapidly publishing 1000 messages in a loop may be constrained by flow
# control.
1000.timesdo|i|
publisher.publish_async "message #{i}"do|result|
raise"Failed to publish the message."unlessresult.succeeded?
end
end
# Stop the async_publisher to send all queued messages immediately.
publisher.async_publisher .stop.wait!
puts"Published messages with flow control settings to #{topic_id}."