Configure Pub/Sub notifications

importcom.google.cloud.pubsub.v1.AckReplyConsumer ;
importcom.google.cloud.pubsub.v1.MessageReceiver ;
importcom.google.cloud.pubsub.v1.Subscriber ;
importcom.google.cloud.pubsub.v1.SubscriptionAdminClient ;
importcom.google.pubsub.v1.ProjectSubscriptionName ;
importcom.google.pubsub.v1.PubsubMessage ;
importcom.google.pubsub.v1.PushConfig ;
importcom.google.pubsub.v1.Subscription ;
importcom.google.pubsub.v1.SubscriptionName ;
importcom.google.pubsub.v1.TopicName ;
importio.grpc.StatusRuntimeException;
importjava.io.IOException;
importjava.lang.InterruptedException;
importjava.util.concurrent.TimeUnit;
publicclass Subscriptions{
// Handle incoming Occurrences using a Cloud Pub/Sub subscription
publicstaticintpubSub(StringsubId,longtimeoutSeconds,StringprojectId)
throwsInterruptedException{
// String subId = "my-occurrence-subscription";
// long timeoutSeconds = 20;
// String projectId = "my-project-id";
Subscriber subscriber=null;
MessageReceiverExamplereceiver=newMessageReceiverExample();
try{
// Subscribe to the requested Pub/Sub channel
ProjectSubscriptionName subName=ProjectSubscriptionName .of(projectId,subId);
subscriber=Subscriber .newBuilder(subName,receiver).build();
subscriber.startAsync ().awaitRunning();
// Sleep to listen for messages
TimeUnit.SECONDS.sleep(timeoutSeconds);
}finally{
// Stop listening to the channel
if(subscriber!=null){
subscriber.stopAsync();
}
}
// Print and return the number of Pub/Sub messages received
System.out.println(receiver.messageCount);
returnreceiver.messageCount;
}
// Custom class to handle incoming Pub/Sub messages
// In this case, the class will simply log and count each message as it comes in
staticclass MessageReceiverExampleimplementsMessageReceiver {
publicintmessageCount=0;
@Override
publicsynchronizedvoidreceiveMessage(PubsubMessage message,AckReplyConsumer consumer){
// Every time a Pub/Sub message comes in, print it and count it
System.out.println("Message "+messageCount+": "+message.getData ().toStringUtf8());
messageCount+=1;
// Acknowledge the message
consumer.ack ();
}
}
// Creates and returns a Pub/Sub subscription object listening to the Occurrence topic
publicstaticSubscription createOccurrenceSubscription(StringsubId,StringprojectId)
throwsIOException,StatusRuntimeException,InterruptedException{
// This topic id will automatically receive messages when Occurrences are added or modified
StringtopicId="container-analysis-occurrences-v1";
TopicName topicName=TopicName .of(projectId,topicId);
SubscriptionName subName=SubscriptionName .of(projectId,subId);
SubscriptionAdminClient client=SubscriptionAdminClient .create();
PushConfig config=PushConfig .getDefaultInstance();
Subscription sub=client.createSubscription (subName,topicName,config,0);
returnsub;
}
}

import(
"context"
"fmt"
"io"
"sync"
"time"
pubsub"cloud.google.com/go/pubsub"
)
// occurrencePubsub handles incoming Occurrences using a Cloud Pub/Sub subscription.
funcoccurrencePubsub(wio.Writer,subscriptionIDstring,timeouttime.Duration,projectIDstring)(int,error){
// subscriptionID := fmt.Sprintf("my-occurrences-subscription")
// timeout := time.Duration(20) * time.Second
ctx:=context.Background()
varmusync.Mutex
client,err:=pubsub.NewClient (ctx,projectID)
iferr!=nil{
return-1,fmt.Errorf("pubsub.NewClient: %w",err)
}
// Subscribe to the requested Pub/Sub channel.
sub:=client.Subscription(subscriptionID)
count:=0
// Listen to messages for 'timeout' seconds.
ctx,cancel:=context.WithTimeout(ctx,timeout)
defercancel()
err=sub.Receive(ctx,func(ctxcontext.Context,msg*pubsub.Message){
mu.Lock()
count=count+1
fmt.Fprintf(w,"Message %d: %q\n",count,string(msg.Data))
msg.Ack()
mu.Unlock()
})
iferr!=nil{
return-1,fmt.Errorf("sub.Receive: %w",err)
}
// Print and return the number of Pub/Sub messages received.
fmt.Fprintln(w,count)
returncount,nil
}
// createOccurrenceSubscription creates a new Pub/Sub subscription object listening to the Occurrence topic.
funccreateOccurrenceSubscription(subscriptionID,projectIDstring)error{
// subscriptionID := fmt.Sprintf("my-occurrences-subscription")
ctx:=context.Background()
client,err:=pubsub.NewClient (ctx,projectID)
iferr!=nil{
returnfmt.Errorf("pubsub.NewClient: %w",err)
}
deferclient.Close()
// This topic id will automatically receive messages when Occurrences are added or modified
topicID:="container-analysis-occurrences-v1"
topic:=client.Topic(topicID)
config:=pubsub.SubscriptionConfig{Topic:topic}
_,err=client.CreateSubscription(ctx,subscriptionID,config)
returnfmt.Errorf("client.CreateSubscription: %w",err)
}

/**
 * TODO(developer): Uncomment these variables before running the sample
 */
// const projectId = 'your-project-id', // Your GCP Project ID
// const subscriptionId = 'my-sub-id', // A user-specified subscription to the 'container-analysis-occurrences-v1' topic
// const timeoutSeconds = 30 // The number of seconds to listen for the new Pub/Sub Messages
// Import the pubsub library and create a client, topic and subscription
const{PubSub}=require('@google-cloud/pubsub');
constpubsub=newPubSub ({projectId});
constsubscription=pubsub.subscription(subscriptionId);
// Handle incoming Occurrences using a Cloud Pub/Sub subscription
letcount=0;
constmessageHandler=message=>{
count++;
message.ack ();
};
// Listen for new messages until timeout is hit
subscription .on ('message',messageHandler);
setTimeout(()=>{
subscription.removeListener('message',messageHandler);
console.log(`Polled ${count} occurrences`);
},timeoutSeconds*1000);

# subscription_id = "A user-specified identifier for the new subscription"
# timeout_seconds = "The number of seconds to listen for new Pub/Sub messages"
# project_id = "Your Google Cloud project ID"
require"google/cloud/pubsub"
pubsub=Google::Cloud::PubSub .new project_id:project_id
subscription_admin=pubsub.subscription_admin
subscription=subscription_admin.create_subscription\
name:pubsub.subscription_path(subscription_id),
topic:pubsub.topic_path("container-analysis-occurrences-v1")
subscriber=pubsub.subscriber subscription.name
count=0
listener=subscriber.listen do|received_message|
count+=1
# Process incoming occurrence here
puts"Message #{count}: #{received_message.data}"
received_message.acknowledge!
end
listener.start
# Wait for incoming occurrences
sleeptimeout_seconds
listener.stop.wait!
subscription_admin.delete_subscriptionsubscription:subscription.name
# Print and return the total number of Pub/Sub messages received
puts"Total Messages Received: #{count}"
count

importtime
fromgoogle.api_core.exceptionsimport AlreadyExists
fromgoogle.cloud.pubsubimport SubscriberClient
fromgoogle.cloud.pubsub_v1.subscriber.messageimport Message
defpubsub(subscription_id: str, timeout_seconds: int, project_id: str) -> int:
"""Respond to incoming occurrences using a Cloud Pub/Sub subscription."""
 # subscription_id := 'my-occurrences-subscription'
 # timeout_seconds = 20
 # project_id = 'my-gcp-project'
 client = SubscriberClient()
 subscription_name = client.subscription_path(project_id, subscription_id)
 receiver = MessageReceiver()
 client.subscribe (subscription_name, receiver.pubsub_callback)
 # listen for 'timeout' seconds
 for _ in range(timeout_seconds):
 time.sleep(1)
 # print and return the number of pubsub messages received
 print(receiver.msg_count)
 return receiver.msg_count
classMessageReceiver:
"""Custom class to handle incoming Pub/Sub messages."""
 def__init__(self) -> None:
 # initialize counter to 0 on initialization
 self.msg_count = 0
 defpubsub_callback(self, message: Message) -> None:
 # every time a pubsub message comes in, print it and count it
 self.msg_count += 1
 print(f"Message {self.msg_count}: {message.data }")
 message.ack ()
defcreate_occurrence_subscription(subscription_id: str, project_id: str) -> bool:
"""Creates a new Pub/Sub subscription object listening to the
 Container Analysis Occurrences topic."""
 # subscription_id := 'my-occurrences-subscription'
 # project_id = 'my-gcp-project'
 topic_id = "container-analysis-occurrences-v1"
 client = SubscriberClient()
 topic_name = f"projects/{project_id}/topics/{topic_id}"
 subscription_name = client.subscription_path(project_id, subscription_id)
 success = True
 try:
 client.create_subscription({"name": subscription_name, "topic": topic_name})
 except AlreadyExists:
 # if subscription already exists, do nothing
 pass
 else:
 success = False
 return success