sarama/kafka/client.go

package kafka

import k "sarama/protocol"

type Client struct {
	id    string
	cache *metadataCache
}

func NewClient(id string, host string, port int32) (client *Client, err error) {
	client = new(Client)
	client.id = id
	client.cache, err = newMetadataCache(client, host, port)
	if err != nil {
		return nil, err
	}
	return client, nil
}

func (client *Client) leader(topic string, partition_id int32) (*k.Broker, error) {
	leader := client.cache.leader(topic, partition_id)

	if leader == nil {
		err := client.cache.refreshTopic(topic)
		if err != nil {
			return nil, err
		}

		leader = client.cache.leader(topic, partition_id)
	}

	if leader == nil {
		return nil, k.UNKNOWN_TOPIC_OR_PARTITION
	}

	return leader, nil
}

func (client *Client) partitions(topic string) ([]int32, error) {
	partitions := client.cache.partitions(topic)

	if partitions == nil {
		err := client.cache.refreshTopic(topic)
		if err != nil {
			return nil, err
		}

		partitions = client.cache.partitions(topic)
	}

	if partitions == nil {
		return nil, NoSuchTopic
	}

	return partitions, nil
}