(ctx context.Context, dt driver.Topic, testName string)
| 95 | } |
| 96 | |
| 97 | func (h *harness) CreateSubscription(ctx context.Context, dt driver.Topic, testName string) (ds driver.Subscription, cleanup func(), err error) { |
| 98 | // We may encounter subscriptions that were created by a previous test run |
| 99 | // and were not properly cleaned up. In such a case delete the existing |
| 100 | // subscription and create a new subscription with a higher subscription |
| 101 | // number (to avoid cool-off issues between deletion and re-creation). |
| 102 | for { |
| 103 | subName := fmt.Sprintf("%s-subscription-%d", sanitize(testName), atomic.AddUint32(&h.numSubs, 1)) |
| 104 | subPath := fmt.Sprintf("projects/%s/subscriptions/%s", projectID, subName) |
| 105 | ds, cleanup, err := createSubscription(ctx, h.subClient, dt, subName, subPath) |
| 106 | if err != nil && status.Code(err) == codes.AlreadyExists { |
| 107 | // Delete the subscription and retry. |
| 108 | h.subClient.DeleteSubscription(ctx, &pubsubpb.DeleteSubscriptionRequest{Subscription: subPath}) |
| 109 | continue |
| 110 | } |
| 111 | return ds, cleanup, err |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | func createSubscription(ctx context.Context, subClient *raw.SubscriberClient, dt driver.Topic, subName, subPath string) (ds driver.Subscription, cleanup func(), err error) { |
| 116 | t := dt.(*topic) |
nothing calls this directly
no test coverage detected