func (m *metadataCache) getAddrs() []string {
	m.mu.RLock()
	var (
		metadata       = m.metadata
		bootstrapAddrs = m.bootstrapAddrs
	)
	m.mu.RUnlock()
	addrs := metadata.Addrs()
	for _, addr := range bootstrapAddrs {
		addrs = append(addrs, addr)
	}
	return addrs
}
getAddrs returns a list of all broker addresses.
package liftbridge
import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"
	proto "github.com/liftbridge-io/liftbridge-api/go"
)
type StreamInfo struct {
	subject    string
	name       string
	partitions map[int32]*PartitionInfo
}
func (s *StreamInfo) GetPartition(id int32) *PartitionInfo {
	return s.partitions[id]
}
func (s *StreamInfo) Partitions() map[int32]*PartitionInfo {
	return s.partitions
}
type PartitionInfo struct {
	id       int32
	leader   *BrokerInfo
	replicas []*BrokerInfo
	isr      []*BrokerInfo
	paused   bool
}
func (p *PartitionInfo) ID() int32 {
	return p.id
}
func (p *PartitionInfo) Replicas() []*BrokerInfo {
	return p.replicas
}
func (p *PartitionInfo) ISR() []*BrokerInfo {
	return p.isr
}
func (p *PartitionInfo) Leader() *BrokerInfo {
	return p.leader
}
func (p *PartitionInfo) Paused() bool {
	return p.paused
}
type BrokerInfo struct {
	id   string
	host string
	port int32
}
func (b *BrokerInfo) ID() string {
	return b.id
}
func (b *BrokerInfo) Host() string {
	return b.host
}
func (b *BrokerInfo) Port() int32 {
	return b.port
}
func (b *BrokerInfo) Addr() string {
	return fmt.Sprintf("%s:%d", b.host, b.port)
}
type Metadata struct {
	lastUpdated time.Time
	brokers     map[string]*BrokerInfo
	addrs       map[string]struct{}
	streams     map[string]*StreamInfo
}
func newMetadata(brokers map[string]*BrokerInfo, streams map[string]*StreamInfo) *Metadata {
	addrs := make(map[string]struct{}, len(brokers))
	for _, broker := range brokers {
		addrs[broker.Addr()] = struct{}{}
	}
	return &Metadata{
		lastUpdated: time.Now(),
		brokers:     brokers,
		addrs:       addrs,
		streams:     streams,
	}
}
func (m *Metadata) LastUpdated() time.Time {
	return m.lastUpdated
}
func (m *Metadata) Brokers() []*BrokerInfo {
	brokers := make([]*BrokerInfo, 0, len(m.brokers))
	for _, broker := range m.brokers {
		brokers = append(brokers, broker)
	}
	return brokers
}
func (m *Metadata) Addrs() []string {
	addrs := make([]string, 0, len(m.addrs))
	for addr := range m.addrs {
		addrs = append(addrs, addr)
	}
	return addrs
}
func (m *Metadata) GetStream(name string) *StreamInfo {
	return m.streams[name]
}
func (m *Metadata) Streams() []*StreamInfo {
	streams := make([]*StreamInfo, 0, len(m.streams))
	for _, stream := range m.streams {
		streams = append(streams, stream)
	}
	return streams
}
func (m *Metadata) PartitionCountForStream(stream string) int32 {
	info := m.GetStream(stream)
	if info == nil {
		return 0
	}
	return int32(len(info.partitions))
}
func (m *Metadata) hasStreamMetadata(stream string) bool {
	return m.GetStream(stream) != nil
}
type metadataCache struct {
	mu             sync.RWMutex
	metadata       *Metadata
	bootstrapAddrs []string
	doRPC          func(func(proto.APIClient) error) error
}
func newMetadataCache(addrs []string, doRPC func(func(proto.APIClient) error) error) *metadataCache {
	return &metadataCache{
		metadata:       &Metadata{},
		bootstrapAddrs: addrs,
		doRPC:          doRPC,
	}
}
func (m *metadataCache) update(ctx context.Context) (*Metadata, error) {
	var resp *proto.FetchMetadataResponse
	if err := m.doRPC(func(client proto.APIClient) (err error) {
		resp, err = client.FetchMetadata(ctx, &proto.FetchMetadataRequest{})
		return err
	}); err != nil {
		return nil, err
	}
	brokers := make(map[string]*BrokerInfo, len(resp.Brokers))
	for _, broker := range resp.Brokers {
		brokers[broker.Id] = &BrokerInfo{
			id:   broker.Id,
			host: broker.Host,
			port: broker.Port,
		}
	}
	streams := make(map[string]*StreamInfo)
	for _, streamMetadata := range resp.Metadata {
		stream := &StreamInfo{
			subject:    streamMetadata.Subject,
			name:       streamMetadata.Name,
			partitions: make(map[int32]*PartitionInfo, len(streamMetadata.Partitions)),
		}
		for _, partition := range streamMetadata.Partitions {
			replicas := make([]*BrokerInfo, 0, len(partition.Replicas))
			for _, replica := range partition.Replicas {
				replicas = append(replicas, brokers[replica])
			}
			isr := make([]*BrokerInfo, 0, len(partition.Isr))
			for _, replica := range partition.Isr {
				isr = append(isr, brokers[replica])
			}
			stream.partitions[partition.Id] = &PartitionInfo{
				id:       partition.Id,
				leader:   brokers[partition.Leader],
				replicas: replicas,
				isr:      isr,
				paused:   partition.Paused,
			}
		}
		streams[stream.name] = stream
	}
	metadata := newMetadata(brokers, streams)
	m.mu.Lock()
	m.metadata = metadata
	m.mu.Unlock()
	return metadata, nil
}