kafka

var OpenSubscription = func(ctx context.Context, topic string) (interface{}, error) {
	subscriptionGroupsSync.Lock()
	defer subscriptionGroupsSync.Unlock()
	groupID := subscriptionGroups[topic]
	subscriptionGroups[topic] = groupID + 1

	uri := fmt.Sprintf(subscribeURI, groupID, topic)
	return pubsub.OpenSubscription(ctx, uri)
}

var OpenTopic = func(ctx context.Context, topic string) (interface{}, error) {
	return pubsub.OpenTopic(ctx, openTopicURI+topic)
}