googlecloud subscriber: replace ctx with configurable timeout

message/infrastructure/googlecloud/pubsub_bench_test.go

@@ -1,9 +1,7 @@
 package googlecloud_test
 
 import (
-	"context"
 	"testing"
-	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
 	"github.com/ThreeDotsLabs/watermill/message"
@@ -22,11 +20,7 @@ func BenchmarkSubscriber(b *testing.B) {
 			panic(err)
 		}
 
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-		defer cancel()
-
 		subscriber, err := googlecloud.NewSubscriber(
-			ctx,
 			googlecloud.SubscriberConfig{},
 			logger,
 		)

message/infrastructure/googlecloud/pubsub_test.go

@@ -30,11 +30,7 @@ func newPubSub(t *testing.T, marshaler googlecloud.MarshalerUnmarshaler, subscri
 	)
 	require.NoError(t, err)
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-	defer cancel()
-
 	subscriber, err := googlecloud.NewSubscriber(
-		ctx,
 		googlecloud.SubscriberConfig{
 			GenerateSubscriptionName: subscriptionName,
 			SubscriptionConfig: pubsub.SubscriptionConfig{
@@ -74,9 +70,6 @@ func TestPublishSubscribe(t *testing.T) {
 }
 
 func TestSubscriberUnexpectedTopicForSubscription(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-
 	rand.Seed(time.Now().Unix())
 	testNumber := rand.Int()
 	logger := watermill.NewStdLogger(true, true)
@@ -85,21 +78,24 @@ func TestSubscriberUnexpectedTopicForSubscription(t *testing.T) {
 		return fmt.Sprintf("sub_%d", testNumber)
 	}
 
-	sub1, err := googlecloud.NewSubscriber(ctx, googlecloud.SubscriberConfig{
+	sub1, err := googlecloud.NewSubscriber(googlecloud.SubscriberConfig{
 		GenerateSubscriptionName: subNameFn,
 	}, logger)
 	require.NoError(t, err)
 
 	topic1 := fmt.Sprintf("topic1_%d", testNumber)
 
-	sub2, err := googlecloud.NewSubscriber(ctx, googlecloud.SubscriberConfig{
+	sub2, err := googlecloud.NewSubscriber(googlecloud.SubscriberConfig{
 		GenerateSubscriptionName: subNameFn,
 	}, logger)
 	require.NoError(t, err)
 	topic2 := fmt.Sprintf("topic2_%d", testNumber)
 
 	howManyMessages := 100
 
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
 	messagesTopic1, err := sub1.Subscribe(ctx, topic1)
 	require.NoError(t, err)
 

message/infrastructure/googlecloud/subscriber.go

@@ -65,6 +65,8 @@ type SubscriberConfig struct {
 	// Otherwise, trying to create a subscription on non-existent topic results in `ErrTopicDoesNotExist`.
 	DoNotCreateTopicIfMissing bool
 
+	// ConnectTimeout defines the timeout for connecting to Pub/Sub
+	ConnectTimeout time.Duration
 	// InitializeTimeout defines the timeout for initializing topics.
 	InitializeTimeout time.Duration
 
@@ -96,6 +98,9 @@ func (c *SubscriberConfig) setDefaults() {
 	if c.GenerateSubscriptionName == nil {
 		c.GenerateSubscriptionName = TopicSubscriptionName
 	}
+	if c.ConnectTimeout == 0 {
+		c.ConnectTimeout = time.Second * 10
+	}
 	if c.InitializeTimeout == 0 {
 		c.InitializeTimeout = time.Second * 10
 	}
@@ -105,12 +110,14 @@ func (c *SubscriberConfig) setDefaults() {
 }
 
 func NewSubscriber(
-	ctx context.Context,
 	config SubscriberConfig,
 	logger watermill.LoggerAdapter,
 ) (*Subscriber, error) {
 	config.setDefaults()
 
+	ctx, cancel := context.WithTimeout(context.Background(), config.ConnectTimeout)
+	defer cancel()
+
 	client, err := pubsub.NewClient(ctx, config.ProjectID, config.ClientOptions...)
 	if err != nil {
 		return nil, err

tools/mill/cmd/googlecloud.go

@@ -40,11 +40,7 @@ For the configuration of consuming/producing of the messages, check the help of
 				}
 			}
 
-			ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
-			defer cancel()
-
 			consumer, err = googlecloud.NewSubscriber(
-				ctx,
 				googlecloud.SubscriberConfig{
 					GenerateSubscriptionName: func(topic string) string {
 						return subName