(ctx context.Context, client *raw.Client, dt driver.Topic, subName string)
| 102 | } |
| 103 | |
| 104 | func createSubscription(ctx context.Context, client *raw.Client, dt driver.Topic, subName string) (ds driver.Subscription, cleanup func(), err error) { |
| 105 | t := dt.(*topic) |
| 106 | _, err = client.SubscriptionAdminClient.CreateSubscription(ctx, &pubsubpb.Subscription{ |
| 107 | Name: subscriptionPath(subName), |
| 108 | Topic: t.publisher.String(), |
| 109 | }) |
| 110 | // We may encounter subscriptions that were created by a previous test run |
| 111 | // and were not properly cleaned up. In such a case delete the existing |
| 112 | // subscription and create a new subscription with a higher subscription |
| 113 | // number (to avoid cool-off issues between deletion and re-creation). |
| 114 | if err != nil && status.Code(err) == codes.AlreadyExists { |
| 115 | deleteSubscription(ctx, client, subName) |
| 116 | return createSubscription(ctx, client, dt, subName) |
| 117 | } |
| 118 | if err != nil { |
| 119 | return nil, nil, fmt.Errorf("failed to create subscription: %w", err) |
| 120 | } |
| 121 | ds = openSubscription(client.Subscriber(subName), nil) |
| 122 | cleanup = func() { |
| 123 | deleteSubscription(ctx, client, subName) |
| 124 | } |
| 125 | return ds, cleanup, nil |
| 126 | } |
| 127 | |
| 128 | func deleteSubscription(ctx context.Context, client *raw.Client, subName string) { |
| 129 | _ = client.SubscriptionAdminClient.DeleteSubscription(ctx, &pubsubpb.DeleteSubscriptionRequest{ |
no test coverage detected