Fix

Author: m110

pkg/googlecloud/publisher.go

@@ -2,6 +2,7 @@ package googlecloud
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"time"
 
@@ -233,7 +234,7 @@ func (p *Publisher) publisher(ctx context.Context, topic string) (pub *pubsub.Pu
 		return pub, nil
 	}
 
-	exists, err := topicExists(ctx, p.client, topic)
+	exists, err := topicExists(ctx, p.client, p.config.ProjectID, topic)
 	if err != nil {
 		return nil, errors.Wrapf(err, "could not check if topic %s exists", topic)
 	}
@@ -247,7 +248,7 @@ func (p *Publisher) publisher(ctx context.Context, topic string) (pub *pubsub.Pu
 	}
 
 	_, err = p.client.TopicAdminClient.CreateTopic(ctx, &pubsubpb.Topic{
-		Name: topic,
+		Name: fullyQualifiedTopicName(p.config.ProjectID, topic),
 	})
 	if err != nil {
 		return nil, errors.Wrapf(err, "could not create topic %s", topic)
@@ -256,9 +257,9 @@ func (p *Publisher) publisher(ctx context.Context, topic string) (pub *pubsub.Pu
 	return pub, nil
 }
 
-func topicExists(ctx context.Context, client *pubsub.Client, topic string) (bool, error) {
+func topicExists(ctx context.Context, client *pubsub.Client, projectID string, topic string) (bool, error) {
 	_, err := client.TopicAdminClient.GetTopic(ctx, &pubsubpb.GetTopicRequest{
-		Topic: topic,
+		Topic: fullyQualifiedTopicName(projectID, topic),
 	})
 	if err != nil {
 		if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
@@ -268,3 +269,7 @@ func topicExists(ctx context.Context, client *pubsub.Client, topic string) (bool
 	}
 	return true, nil
 }
+
+func fullyQualifiedTopicName(projectID string, topic string) string {
+	return fmt.Sprintf("projects/%s/topics/%s", projectID, topic)
+}

pkg/googlecloud/pubsub_test.go

@@ -136,13 +136,15 @@ func TestSubscriberAllowedWhenAttachedToAnotherTopic(t *testing.T) {
 	}
 
 	sub1, err := googlecloud.NewSubscriber(googlecloud.SubscriberConfig{
+		ProjectID:                "tests",
 		GenerateSubscriptionName: subNameFn,
 	}, logger)
 	require.NoError(t, err)
 
 	topic1 := fmt.Sprintf("topic1_%d", testNumber)
 
 	sub2, err := googlecloud.NewSubscriber(googlecloud.SubscriberConfig{
+		ProjectID:                               "tests",
 		GenerateSubscriptionName:                subNameFn,
 		DoNotEnforceSubscriptionAttachedToTopic: true,
 	}, logger)

pkg/googlecloud/subscriber.go

@@ -384,7 +384,7 @@ func (s *Subscriber) subscription(ctx context.Context, subscriptionName, topicNa
 	sub = s.client.Subscriber(subscriptionName)
 
 	subResp, err := s.client.SubscriptionAdminClient.GetSubscription(ctx, &pubsubpb.GetSubscriptionRequest{
-		Subscription: subscriptionName,
+		Subscription: fullyQualifiedSubscriptionName(s.config.ProjectID, subscriptionName),
 	})
 	if err == nil {
 		return s.existingSubscriber(sub, subResp, topicName)
@@ -400,7 +400,7 @@ func (s *Subscriber) subscription(ctx context.Context, subscriptionName, topicNa
 		return nil, errors.Wrap(ErrSubscriptionDoesNotExist, subscriptionName)
 	}
 
-	tExists, err := topicExists(ctx, s.client, topicName)
+	tExists, err := topicExists(ctx, s.client, s.config.ProjectID, topicName)
 	if err != nil {
 		return nil, errors.Wrapf(err, "could not check if topic %s exists", topicName)
 	}
@@ -411,7 +411,7 @@ func (s *Subscriber) subscription(ctx context.Context, subscriptionName, topicNa
 
 	if !tExists {
 		_, err = s.client.TopicAdminClient.CreateTopic(ctx, &pubsubpb.Topic{
-			Name: topicName,
+			Name: fullyQualifiedTopicName(s.config.topicProjectID(), topicName),
 		})
 
 		if status.Code(err) == codes.AlreadyExists {
@@ -422,8 +422,8 @@ func (s *Subscriber) subscription(ctx context.Context, subscriptionName, topicNa
 	}
 
 	subConfig := s.config.GenerateSubscription(GenerateSubscriptionParams{})
-	subConfig.Name = subscriptionName
-	subConfig.Topic = topicName
+	subConfig.Name = fullyQualifiedSubscriptionName(s.config.ProjectID, subscriptionName)
+	subConfig.Topic = fullyQualifiedTopicName(s.config.topicProjectID(), topicName)
 
 	_, err = s.client.SubscriptionAdminClient.CreateSubscription(ctx, subConfig)
 	if status.Code(err) == codes.AlreadyExists {
@@ -462,12 +462,12 @@ func (s *Subscriber) existingSubscriber(sub *pubsub.Subscriber, subscription *pu
 		return sub, nil
 	}
 
-	fullyQualifiedTopicName := fmt.Sprintf("projects/%s/topics/%s", s.config.topicProjectID(), topic)
+	fullName := fullyQualifiedTopicName(s.config.topicProjectID(), topic)
 
-	if subscription.Topic != fullyQualifiedTopicName {
+	if subscription.Topic != fullName {
 		return nil, errors.Wrap(
 			ErrUnexpectedTopic,
-			fmt.Sprintf("topic of existing sub: %s; expecting: %s", subscription.Topic, fullyQualifiedTopicName),
+			fmt.Sprintf("topic of existing sub: %s; expecting: %s", subscription.Topic, fullName),
 		)
 	}
 
@@ -578,3 +578,7 @@ func subscriptionFromSubscriptionConfig(cfg pubsubv1.SubscriptionConfig) *pubsub
 		MessageTransforms:             messageTransforms,
 	}
 }
+
+func fullyQualifiedSubscriptionName(projectID, subscriptionName string) string {
+	return fmt.Sprintf("projects/%s/subscriptions/%s", projectID, subscriptionName)
+}