v1 Subscribe with exactly once delivery (DEPRECATED)

  import 
  
 ( 
  
 "context" 
  
 "fmt" 
  
 "io" 
  
 "time" 
  
 "cloud.google.com/go/pubsub" 
  
 "google.golang.org/api/option" 
 ) 
 // receiveMessagesWithExactlyOnceDeliveryEnabled instantiates a subscriber client. 
 // This differs from regular subscribing since you must call msg.AckWithResult() 
 // or msg.NackWithResult() instead of the regular Ack/Nack methods. 
 // When exactly once delivery is enabled on the subscription, the message is 
 // guaranteed to not be delivered again if the ack result succeeds. 
 func 
  
 receiveMessagesWithExactlyOnceDeliveryEnabled 
 ( 
 w 
  
 io 
 . 
 Writer 
 , 
  
 projectID 
 , 
  
 subID 
  
 string 
 ) 
  
 error 
  
 { 
  
 // projectID := "my-project-id" 
  
 // subID := "my-sub" 
  
 ctx 
  
 := 
  
 context 
 . 
 Background 
 () 
  
 // Pub/Sub's exactly once delivery guarantee only applies when subscribers connect to the service in the same region. 
  
 // For list of locational endpoints for Pub/Sub, see https://cloud.google.com/pubsub/docs/reference/service_apis_overview#list_of_locational_endpoints 
  
 client 
 , 
  
 err 
  
 := 
  
 pubsub 
 . 
  NewClient 
 
 ( 
 ctx 
 , 
  
 projectID 
 , 
  
 option 
 . 
 WithEndpoint 
 ( 
 "us-west1-pubsub.googleapis.com:443" 
 )) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "pubsub.NewClient: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 defer 
  
 client 
 . 
 Close 
 () 
  
 sub 
  
 := 
  
 client 
 . 
 Subscription 
 ( 
 subID 
 ) 
  
 // Set MinExtensionPeriod high to avoid any unintentional 
  
 // acknowledgment expirations (e.g. due to network events). 
  
 // This can lead to high tail latency in case of client crashes. 
  
 sub 
 . 
 ReceiveSettings 
 . 
 MinExtensionPeriod 
  
 = 
  
 600 
  
 * 
  
 time 
 . 
 Second 
  
 // 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 
 ) 
  
 defer 
  
 cancel 
 () 
  
 err 
  
 = 
  
 sub 
 . 
 Receive 
 ( 
 ctx 
 , 
  
 func 
 ( 
 ctx 
  
 context 
 . 
 Context 
 , 
  
 msg 
  
 * 
 pubsub 
 . 
 Message 
 ) 
  
 { 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Got message: %q\n" 
 , 
  
 string 
 ( 
 msg 
 . 
 Data 
 )) 
  
 r 
  
 := 
  
 msg 
 . 
 AckWithResult 
 () 
  
 // Block until the result is returned and a pubsub.AcknowledgeStatus 
  
 // is returned for the acked message. 
  
 status 
 , 
  
 err 
  
 := 
  
 r 
 . 
 Get 
 ( 
 ctx 
 ) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "MessageID: %s failed when calling result.Get: %v" 
 , 
  
 msg 
 . 
 ID 
 , 
  
 err 
 ) 
  
 } 
  
 switch 
  
 status 
  
 { 
  
 case 
  
 pubsub 
 . 
  AcknowledgeStatusSuccess 
 
 : 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Message successfully acked: %s" 
 , 
  
 msg 
 . 
 ID 
 ) 
  
 case 
  
 pubsub 
 . 
  AcknowledgeStatusInvalidAckID 
 
 : 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Message failed to ack with response of Invalid. ID: %s" 
 , 
  
 msg 
 . 
 ID 
 ) 
  
 case 
  
 pubsub 
 . 
  AcknowledgeStatusPermissionDenied 
 
 : 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Message failed to ack with response of Permission Denied. ID: %s" 
 , 
  
 msg 
 . 
 ID 
 ) 
  
 case 
  
 pubsub 
 . 
  AcknowledgeStatusFailedPrecondition 
 
 : 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Message failed to ack with response of Failed Precondition. ID: %s" 
 , 
  
 msg 
 . 
 ID 
 ) 
  
 case 
  
 pubsub 
 . 
  AcknowledgeStatusOther 
 
 : 
  
 fmt 
 . 
 Fprintf 
 ( 
 w 
 , 
  
 "Message failed to ack with response of Other. ID: %s" 
 , 
  
 msg 
 . 
 ID 
 ) 
  
 default 
 : 
  
 } 
  
 }) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 fmt 
 . 
 Errorf 
 ( 
 "got err from sub.Receive: %w" 
 , 
  
 err 
 ) 
  
 } 
  
 return 
  
 nil 
 } 
 

  require 
  
 "google/cloud/pubsub" 
 # Shows how to register callback to acknowledge method and access the result passed in 
 class 
  
 PubsubSubscriberExactlyOnceDelivery 
  
 def 
  
 subscriber_exactly_once_delivery 
  
 project_id 
 :, 
  
 topic_id 
 :, 
  
 subscription_id 
 : 
  
 pubsub 
  
 = 
  
 Google 
 :: 
 Cloud 
 :: 
  Pubsub 
 
 . 
 new 
  
 project_id 
 : 
  
 project_id 
  
 topic 
  
 = 
  
 pubsub 
 . 
 topic 
  
 topic_id 
  
 subscription 
  
 = 
  
 pubsub 
 . 
 subscription 
  
 subscription_id 
  
 subscriber 
  
 = 
  
 subscription 
 . 
 listen 
  
 do 
  
 | 
 received_message 
 | 
  
 puts 
  
 "Received message: 
 #{ 
 received_message 
 . 
 data 
 } 
 " 
  
 # Pass in callback to access the acknowledge result. 
  
 # For subscription with Exactly once delivery disabled the result will be success always. 
  
 received_message 
 . 
 acknowledge! 
  
 do 
  
 | 
 result 
 | 
  
 puts 
  
 "Acknowledge result's status: 
 #{ 
 result 
 . 
 status 
 } 
 " 
  
 end 
  
 end 
  
 subscriber 
 . 
 start 
  
 # Let the main thread sleep for 60 seconds so the thread for listening 
  
 # messages does not quit 
  
 sleep 
  
 60 
  
 subscriber 
 . 
 stop 
 . 
 wait! 
  
 end 
  
 def 
  
 self 
 . 
 run 
  
 # TODO(developer): Replace these variables before running the sample. 
  
 project_id 
  
 = 
  
 "your-project-id" 
  
 topic_id 
  
 = 
  
 "your-topic-id" 
  
 subscription_id 
  
 = 
  
 "id-for-new-subcription" 
  
 # subscription with exactly once delivery enabled 
  
 PubsubSubscriberExactlyOnceDelivery 
 . 
 new 
 . 
 subscriber_exactly_once_delivery 
  
 project_id 
 : 
  
 project_id 
 , 
  
 topic_id 
 : 
  
 topic_id 
 , 
  
 subscription_id 
 : 
  
 subscription_id 
  
 end 
 end 
 if 
  
 $PROGRAM_NAME 
  
 == 
  
 __FILE__ 
  
 PubsubSubscriberExactlyOnceDelivery 
 . 
 run 
 end