func TestReceiveCreateSubscriptionAfterFailure(t *testing.T) {
	for _, failureMethod := range []string{
		"/google.pubsub.v1.Publisher/GetTopic",
		"/google.pubsub.v1.Publisher/CreateTopic",
		"/google.pubsub.v1.Subscriber/GetSubscription",
		"/google.pubsub.v1.Subscriber/CreateSubscription"} {
		t.Run(failureMethod, func(t *testing.T) {
			ctx := context.Background()
			pc := &testPubsubClient{}
			defer pc.Close()
			projectID, topicID, subID := "test-project", "test-topic", "test-sub"
			failureMap := make(map[string][]failPattern)
			failureMap[failureMethod] = []failPattern{{
				ErrReturn: fmt.Errorf("Injected error"),
				Count:     1,
				Delay:     0}}
			client, err := pc.New(ctx, projectID, failureMap)
			if err != nil {
				t.Fatalf("failed to create pubsub client: %v", err)
			}
			defer client.Close()
			psconn := &Connection{
				AllowCreateSubscription: true,
				AllowCreateTopic:        true,
				Client:                  client,
				ProjectID:               projectID,
				TopicID:                 topicID,
				SubscriptionID:          subID,
			}
			ctx2, cancel := context.WithCancel(ctx)
			errRet := make(chan error)
			go func() {
				errRet <- psconn.Receive(ctx2, func(_ context.Context, msg *pubsub.Message) {
					msg.Ack()
				})
			}()
			select {
			case err := <-errRet:
				if err == nil {
					t.Fatalf("unexpected nil error from Receive")
				}
			case <-time.After(time.Minute):
				cancel()
				t.Fatalf("timeout waiting for receive error")
			}
			errRet2 := make(chan error)
			ctx2, cancel = context.WithCancel(context.Background())
			go func() {
				errRet2 <- psconn.Receive(ctx2, func(_ context.Context, msg *pubsub.Message) {
					msg.Ack()
				})
			}()
			for _, delay := range []time.Duration{time.Second / 4, time.Second, 20 * time.Second, 40 * time.Second} {
				time.Sleep(delay)
				ok, err := client.Subscription(subID).Exists(ctx)
				if ok == true && err == nil {
					break
				}
			}
			select {
			case err := <-errRet2:
				t.Errorf("unexpected error from Receive: %v", err)
			default:
			}
			if ok, err := client.Topic(topicID).Exists(ctx); err != nil || !ok {
				t.Errorf("topic id=%s got exists=%v want=true, err=%v", topicID, ok, err)
			}
			if ok, err := client.Subscription(subID).Exists(ctx); err != nil || !ok {
				t.Errorf("subscription id=%s got exists=%v want=true, err=%v", subID, ok, err)
			}
			cancel()
		})
	}
}
Test that creating a subscription after a failed attempt to create a subsciption works
package internal
import (
	"context"
	"fmt"
	"sync"
	"testing"
	"time"
	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsub/pstest"
	"github.com/google/go-cmp/cmp"
	"google.golang.org/api/option"
	"google.golang.org/grpc"
)
type testPubsubClient struct {
	srv  *pstest.Server
	conn *grpc.ClientConn
}
type failPattern struct {
	ErrReturn error
	Count int
	Delay time.Duration
}
func (pc *testPubsubClient) New(ctx context.Context, projectID string, failureMap map[string][]failPattern) (*pubsub.Client, error) {
	pc.srv = pstest.NewServer()
	var err error
	var conn *grpc.ClientConn
	if len(failureMap) == 0 {
		conn, err = grpc.Dial(pc.srv.Addr, grpc.WithInsecure())
	} else {
		conn, err = grpc.Dial(pc.srv.Addr, grpc.WithInsecure(), grpc.WithUnaryInterceptor(makeFailureIntercept(failureMap)))
	}
	if err != nil {
		return nil, err
	}
	pc.conn = conn
	return pubsub.NewClient(ctx, projectID, option.WithGRPCConn(conn))
}
func (pc *testPubsubClient) Close() {
	pc.srv.Close()
	pc.conn.Close()
}
func makeFailureIntercept(failureMap map[string][]failPattern) grpc.UnaryClientInterceptor {
	var lock sync.Mutex
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		var injectedErr error
		var delay time.Duration
		lock.Lock()
		if failureMap != nil {
			fpArr := failureMap[method]
			if len(fpArr) != 0 {
				injectedErr = fpArr[0].ErrReturn
				delay = fpArr[0].Delay
				if fpArr[0].Count != 0 {
					fpArr[0].Count--
					if fpArr[0].Count == 0 {
						failureMap[method] = fpArr[1:]
					}
				}
			}
		}
		lock.Unlock()
		if delay != 0 {
			time.Sleep(delay)
		}
		if injectedErr != nil {
			return injectedErr
		} else {
			return invoker(ctx, method, req, reply, cc, opts...)
		}
	}
}
func verifyTopicDeleteWorks(t *testing.T, client *pubsub.Client, psconn *Connection, topicID string) {
	ctx := context.Background()
	if ok, err := client.Topic(topicID).Exists(ctx); err != nil || !ok {
		t.Errorf("topic id=%s got exists=%v want=true, err=%v", topicID, ok, err)
	}
	if err := psconn.DeleteTopic(ctx); err != nil {
		t.Errorf("delete topic failed: %v", err)
	}
	if ok, err := client.Topic(topicID).Exists(ctx); err != nil || ok {
		t.Errorf("topic id=%s got exists=%v want=false, err=%v", topicID, ok, err)
	}
}
func verifyTopicDeleteFails(t *testing.T, client *pubsub.Client, psconn *Connection, topicID string) {
	ctx := context.Background()
	if ok, err := client.Topic(topicID).Exists(ctx); err != nil || !ok {
		t.Errorf("topic id=%s got exists=%v want=true, err=%v", topicID, ok, err)
	}
	if err := psconn.DeleteTopic(ctx); err == nil {
		t.Errorf("delete topic succeeded unexpectedly")
	}
	if ok, err := client.Topic(topicID).Exists(ctx); err != nil || !ok {
		t.Errorf("topic id=%s after delete got exists=%v want=true, err=%v", topicID, ok, err)
	}
}
func TestPublishCreateTopic(t *testing.T) {
	ctx := context.Background()
	pc := &testPubsubClient{}
	defer pc.Close()
	projectID, topicID, subID := "test-project", "test-topic", "test-sub"
	client, err := pc.New(ctx, projectID, nil)
	if err != nil {
		t.Fatalf("failed to create pubsub client: %v", err)
	}
	defer client.Close()
	psconn := &Connection{
		AllowCreateSubscription: true,
		AllowCreateTopic:        true,
		Client:                  client,
		ProjectID:               projectID,
		TopicID:                 topicID,
		SubscriptionID:          subID,
	}
	msg := &pubsub.Message{
		ID:   "msg-id-1",
		Data: []byte("msg-data-1"),
	}
	if _, err := psconn.Publish(ctx, msg); err != nil {
		t.Errorf("failed to publish message: %v", err)
	}
	verifyTopicDeleteWorks(t, client, psconn, topicID)
}
func TestPublishExistingTopic(t *testing.T) {
	for _, allowCreate := range []bool{true, false} {
		t.Run(fmt.Sprintf("allowCreate_%v", allowCreate), func(t *testing.T) {
			ctx := context.Background()
			pc := &testPubsubClient{}
			defer pc.Close()
			projectID, topicID, subID := "test-project", "test-topic", "test-sub"
			client, err := pc.New(ctx, projectID, nil)
			if err != nil {
				t.Fatalf("failed to create pubsub client: %v", err)
			}
			defer client.Close()
			psconn := &Connection{
				AllowCreateSubscription: true,
				AllowCreateTopic:        allowCreate,
				Client:                  client,
				ProjectID:               projectID,
				TopicID:                 topicID,
				SubscriptionID:          subID,
			}
			topic, err := client.CreateTopic(ctx, topicID)
			if err != nil {
				t.Fatalf("failed to pre-create topic: %v", err)
			}
			topic.Stop()
			msg := &pubsub.Message{
				ID:   "msg-id-1",
				Data: []byte("msg-data-1"),
			}
			if _, err := psconn.Publish(ctx, msg); err != nil {
				t.Errorf("failed to publish message: %v", err)
			}
			verifyTopicDeleteFails(t, client, psconn, topicID)
		})
	}
}
func TestPublishAfterPublishFailure(t *testing.T) {
	for _, failureMethod := range []string{
		"/google.pubsub.v1.Publisher/GetTopic",
		"/google.pubsub.v1.Publisher/CreateTopic",
		"/google.pubsub.v1.Publisher/Publish"} {
		t.Run(failureMethod, func(t *testing.T) {
			ctx := context.Background()
			pc := &testPubsubClient{}
			defer pc.Close()
			projectID, topicID, subID := "test-project", "test-topic", "test-sub"
			failureMap := make(map[string][]failPattern)
			failureMap[failureMethod] = []failPattern{{
				ErrReturn: fmt.Errorf("Injected error"),
				Count:     1,
				Delay:     0}}
			client, err := pc.New(ctx, projectID, failureMap)
			if err != nil {
				t.Fatalf("failed to create pubsub client: %v", err)
			}
			defer client.Close()
			psconn := &Connection{
				AllowCreateSubscription: true,
				AllowCreateTopic:        true,
				Client:                  client,
				ProjectID:               projectID,
				TopicID:                 topicID,
				SubscriptionID:          subID,
			}
			msg := &pubsub.Message{
				ID:   "msg-id-1",
				Data: []byte("msg-data-1"),
			}
			if _, err := psconn.Publish(ctx, msg); err == nil {
				t.Errorf("Expected publish failure, but didn't see it: %v", err)
			}
			if _, err := psconn.Publish(ctx, msg); err != nil {
				t.Errorf("failed to publish message: %v", err)
			}
			verifyTopicDeleteWorks(t, client, psconn, topicID)
		})
	}
}
func TestPublishCreateTopicAfterDelete(t *testing.T) {
	ctx := context.Background()
	pc := &testPubsubClient{}
	defer pc.Close()
	projectID, topicID, subID := "test-project", "test-topic", "test-sub"
	client, err := pc.New(ctx, projectID, nil)
	if err != nil {
		t.Fatalf("failed to create pubsub client: %v", err)
	}
	defer client.Close()
	psconn := &Connection{
		AllowCreateSubscription: true,
		AllowCreateTopic:        true,
		Client:                  client,
		ProjectID:               projectID,
		TopicID:                 topicID,
		SubscriptionID:          subID,
	}
	msg := &pubsub.Message{
		ID:   "msg-id-1",
		Data: []byte("msg-data-1"),
	}
	if _, err := psconn.Publish(ctx, msg); err != nil {
		t.Errorf("failed to publish message: %v", err)
	}
	verifyTopicDeleteWorks(t, client, psconn, topicID)
	if _, err := psconn.Publish(ctx, msg); err != nil {
		t.Errorf("failed to publish message: %v", err)
	}
	verifyTopicDeleteWorks(t, client, psconn, topicID)
}
func TestPublishCreateTopicNotAllowedFails(t *testing.T) {
	ctx := context.Background()
	pc := &testPubsubClient{}
	defer pc.Close()
	projectID, topicID, subID := "test-project", "test-topic", "test-sub"
	client, err := pc.New(ctx, projectID, nil)
	if err != nil {
		t.Fatalf("failed to create pubsub client: %v", err)
	}
	defer client.Close()
	psconn := &Connection{
		AllowCreateSubscription: true,
		AllowCreateTopic:        false,
		Client:                  client,
		ProjectID:               projectID,
		TopicID:                 topicID,
		SubscriptionID:          subID,
	}
	msg := &pubsub.Message{
		ID:   "msg-id-1",
		Data: []byte("msg-data-1"),
	}
	if _, err := psconn.Publish(ctx, msg); err == nil {
		t.Errorf("publish succeeded unexpectedly")
	}
	if ok, err := client.Topic(topicID).Exists(ctx); err == nil && ok {
		t.Errorf("topic id=%s got exists=%v want=false, err=%v", topicID, ok, err)
	}
}
func TestPublishParallelFailure(t *testing.T) {
	var overallError error
	for _, delay := range []time.Duration{time.Second / 4, 2 * time.Second, 10 * time.Second, 40 * time.Second} {
		overallError = func() error {
			failureMethod := "/google.pubsub.v1.Publisher/GetTopic"
			ctx := context.Background()
			pc := &testPubsubClient{}
			defer pc.Close()
			projectID, topicID, subID := "test-project", "test-topic", "test-sub"
			failureMap := make(map[string][]failPattern)
			failureMap[failureMethod] = []failPattern{{
				ErrReturn: fmt.Errorf("Injected error"),
				Count:     1,
				Delay:     delay}}
			client, err := pc.New(ctx, projectID, failureMap)
			if err != nil {
				t.Fatalf("failed to create pubsub client: %v", err)
			}
			defer client.Close()
			psconn := &Connection{
				AllowCreateSubscription: true,
				AllowCreateTopic:        true,
				Client:                  client,
				ProjectID:               projectID,
				TopicID:                 topicID,
				SubscriptionID:          subID,
			}
			msg := &pubsub.Message{
				ID:   "msg-id-1",
				Data: []byte("msg-data-1"),
			}
			resChan := make(chan error)
			go func() {
				_, err := psconn.Publish(ctx, msg)
				resChan <- err
			}()
			_, errPub1 := psconn.Publish(ctx, msg)
			errPub2 := <-resChan
			if errPub1 == nil || errPub2 == nil {
				return fmt.Errorf("expected dual expected failure, saw (%v) (%v) last run", errPub1, errPub2)
			} else if errPub1 == nil && errPub2 == nil {
				t.Fatalf("Dual success when expecting at least one failure, delay %v", delay)
			}
			return nil
		}()
		if overallError == nil {
			break
		}
	}
	if overallError != nil {
		t.Errorf(overallError.Error())
	}
}
func TestReceiveCreateTopicAndSubscription(t *testing.T) {
	ctx := context.Background()
	pc := &testPubsubClient{}
	defer pc.Close()
	projectID, topicID, subID := "test-project", "test-topic", "test-sub"
	client, err := pc.New(ctx, projectID, nil)
	if err != nil {
		t.Fatalf("failed to create pubsub client: %v", err)
	}
	defer client.Close()
	psconn := &Connection{
		AllowCreateSubscription: true,
		AllowCreateTopic:        true,
		Client:                  client,
		ProjectID:               projectID,
		TopicID:                 topicID,
		SubscriptionID:          subID,
	}
	ctx2, cancel := context.WithCancel(ctx)
	go psconn.Receive(ctx2, func(_ context.Context, msg *pubsub.Message) {
		msg.Ack()
	})
	for _, delay := range []time.Duration{time.Second / 4, time.Second, 20 * time.Second, 40 * time.Second} {
		time.Sleep(delay)
		ok, err := client.Subscription(subID).Exists(ctx)
		if ok == true && err == nil {
			break
		}
	}
	if ok, err := client.Topic(topicID).Exists(ctx); err != nil || !ok {
		t.Errorf("topic id=%s got exists=%v want=true, err=%v", topicID, ok, err)
	}
	if ok, err := client.Subscription(subID).Exists(ctx); err != nil || !ok {
		t.Errorf("subscription id=%s got exists=%v want=true, err=%v", subID, ok, err)
	}
	si, err := psconn.getOrCreateSubscriptionInfo(context.Background(), true)
	if err != nil {
		t.Errorf("error getting subscription info %v", err)
	}
	if si.sub.ReceiveSettings.NumGoroutines != DefaultReceiveSettings.NumGoroutines {
		t.Errorf("subscription receive settings have NumGoroutines=%d, want %d",
			si.sub.ReceiveSettings.NumGoroutines, DefaultReceiveSettings.NumGoroutines)
	}
	cancel()
	if err := psconn.DeleteSubscription(ctx); err != nil {
		t.Errorf("delete subscription failed: %v", err)
	}
	if ok, err := client.Subscription(subID).Exists(ctx); err != nil || ok {
		t.Errorf("subscription id=%s got exists=%v want=false, err=%v", subID, ok, err)
	}
	verifyTopicDeleteWorks(t, client, psconn, topicID)
}
func TestReceiveExistingTopic(t *testing.T) {
	for _, allow := range [](struct{ Sub, Topic bool }){{true, true}, {true, false}, {false, true}, {false, false}} {
		t.Run(fmt.Sprintf("sub_%v__topic_%v", allow.Sub, allow.Topic), func(t *testing.T) {
			ctx := context.Background()
			pc := &testPubsubClient{}
			defer pc.Close()
			projectID, topicID, subID := "test-project", "test-topic", "test-sub"
			client, err := pc.New(ctx, projectID, nil)
			if err != nil {
				t.Fatalf("failed to create pubsub client: %v", err)
			}
			defer client.Close()
			psconn := &Connection{
				AllowCreateSubscription: allow.Sub,
				AllowCreateTopic:        allow.Topic,
				Client:                  client,
				ProjectID:               projectID,
				TopicID:                 topicID,
				SubscriptionID:          subID,
			}
			topic, err := client.CreateTopic(ctx, topicID)
			if err != nil {
				pc.Close()
				t.Fatalf("failed to pre-create topic: %v", err)
			}
			_, err = client.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
				Topic:             topic,
				AckDeadline:       DefaultAckDeadline,
				RetentionDuration: DefaultRetentionDuration,
			})
			topic.Stop()
			if err != nil {
				pc.Close()
				t.Fatalf("failed to pre-createsubscription: %v", err)
			}
			ctx2, cancel := context.WithCancel(ctx)
			go psconn.Receive(ctx2, func(_ context.Context, msg *pubsub.Message) {
				msg.Ack()
			})
			si, err := psconn.getOrCreateSubscriptionInfo(context.Background(), false)
			if err != nil {
				t.Errorf("error getting subscription info %v", err)
			}
			if si.sub.ReceiveSettings.NumGoroutines != DefaultReceiveSettings.NumGoroutines {
				t.Errorf("subscription receive settings have NumGoroutines=%d, want %d",
					si.sub.ReceiveSettings.NumGoroutines, DefaultReceiveSettings.NumGoroutines)
			}
			cancel()
			if err := psconn.DeleteSubscription(ctx); err == nil {
				t.Errorf("delete subscription unexpectedly succeeded")
			}
			if ok, err := client.Subscription(subID).Exists(ctx); err != nil || !ok {
				t.Errorf("subscription id=%s got exists=%v want=true, err=%v", subID, ok, err)
			}
			verifyTopicDeleteFails(t, client, psconn, topicID)
		})
	}
}