<commit_before>package cluster

import (
	"errors"
	"fmt"
	"sync"

	"h12.me/kafka/common"
	"h12.me/kafka/log"
	"h12.me/kafka/proto"
)

var (
	ErrLeaderNotFound = errors.New("leader not found")
	ErrCoordNotFound  = errors.New("coordinator not found")
	ErrNoBrokerFound  = errors.New("no broker found")
)

type (
	C struct {
		topics *topicPartitions
		pool   *brokerPool
		mu     sync.Mutex
	}
	NewBrokerFunc func(addr string) common.Broker
)

func New(newBroker NewBrokerFunc, brokers []string) (*C, error) {
	c := &C{
		topics: newTopicPartitions(),
		pool:   newBrokerPool(newBroker),
	}
	for _, addr := range brokers {
		c.pool.AddAddr(addr)
	}
	return c, nil
}

func (c *C) Partitions(topic string) ([]int32, error) {
	partitions := c.topics.getPartitions(topic)
	if len(partitions) > 0 {
		return partitions, nil
	}
	if err := c.updateFromTopicMetadata(topic); err != nil {
		return nil, err
	}
	partitions = c.topics.getPartitions(topic)
	if len(partitions) > 0 {
		return partitions, nil
	}
	return nil, fmt.Errorf("topic %s not found", topic)
}

func (c *C) Coordinator(group string) (common.Broker, error) {
	if coord, err := c.pool.GetCoordinator(group); err == nil {
		return coord, nil
	}
	if err := c.updateCoordinator(group); err != nil {
		return nil, err
	}
	return c.pool.GetCoordinator(group)
}
<commit_msg>improve constructor of cluster<commit_after>package cluster

import (
	"errors"
	"fmt"
	"sync"

	"h12.me/kafka/common"
	"h12.me/kafka/log"
	"h12.me/kafka/proto"
)

var (
	ErrLeaderNotFound = errors.New("leader not found")
	ErrCoordNotFound  = errors.New("coordinator not found")
	ErrNoBrokerFound  = errors.New("no broker found")
)

type (
	C struct {
		topics *topicPartitions
		pool   *brokerPool
		mu     sync.Mutex
	}
	NewBrokerFunc func(addr string) common.Broker
)

func New(newBroker NewBrokerFunc, brokers []string) common.Cluster {
	c := &C{
		topics: newTopicPartitions(),
		pool:   newBrokerPool(newBroker),
	}
	for _, addr := range brokers {
		c.pool.AddAddr(addr)
	}
	return c
}

func (c *C) Partitions(topic string) ([]int32, error) {
	partitions := c.topics.getPartitions(topic)
	if len(partitions) > 0 {
		return partitions, nil
	}
	if err := c.updateFromTopicMetadata(topic); err != nil {
		return nil, err
	}
	partitions = c.topics.getPartitions(topic)
	if len(partitions) > 0 {
		return partitions, nil
	}
	return nil, fmt.Errorf("topic %s not found", topic)
}

func (c *C) Coordinator(group string) (common.Broker, error) {
	if coord, err := c.pool.GetCoordinator(group); err == nil {
		return coord, nil
	}
	if err := c.updateCoordinator(group); err != nil {
		return nil, err
	}
	return c.pool.GetCoordinator(group)
}

{'fraction_non_alphanumeric': 0.13583489681050656, 'fraction_numerical': 0.0075046904315197, 'mean_word_length': 4.078095238095238, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4181615', 'n_tokens_mistral': 1046, 'n_tokens_neox': 999, 'n_words': 241}