(ctx context.Context, client *raw.Client, topicName string)
| 67 | } |
| 68 | |
| 69 | func createTopic(ctx context.Context, client *raw.Client, topicName string) (dt driver.Topic, cleanup func(), err error) { |
| 70 | _, err = client.TopicAdminClient.CreateTopic(ctx, &pubsubpb.Topic{Name: topicPath(topicName)}) |
| 71 | // We may encounter topics that were created by a previous test run and were |
| 72 | // not properly cleaned up. In such a case delete the existing topic and create |
| 73 | // a new topic with a higher topic number (to avoid cool-off issues between |
| 74 | // deletion and re-creation). |
| 75 | if err != nil && status.Code(err) == codes.AlreadyExists { |
| 76 | deleteTopic(ctx, client, topicName) |
| 77 | return createTopic(ctx, client, topicName) |
| 78 | } |
| 79 | if err != nil { |
| 80 | return nil, nil, fmt.Errorf("failed to create topic: %w", err) |
| 81 | } |
| 82 | dt = openTopic(client.Publisher(topicName)) |
| 83 | cleanup = func() { |
| 84 | deleteTopic(ctx, client, topicName) |
| 85 | } |
| 86 | return dt, cleanup, nil |
| 87 | } |
| 88 | |
| 89 | func deleteTopic(ctx context.Context, client *raw.Client, topicName string) { |
| 90 | _ = client.TopicAdminClient.DeleteTopic(ctx, &pubsubpb.DeleteTopicRequest{ |
no test coverage detected