func KeepAliveTime(keepAlive time.Duration) ClientOption {
	return func(o *ClientOptions) error {
		o.KeepAliveTime = keepAlive
		return nil
	}
}
KeepAliveTime is a ClientOption to set the amount of time a pooled
connection can be idle before it is closed and removed from the pool. The
default is 30 seconds.
package liftbridge
import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/status"
	proto "github.com/liftbridge-io/liftbridge-api/go"
)
const MaxReplicationFactor int32 = -1
type StartPosition int32
func (s StartPosition) toProto() proto.StartPosition {
	return proto.StartPosition(s)
}
const (
	defaultMaxConnsPerBroker   = 2
	defaultKeepAliveTime       = 30 * time.Second
	defaultResubscribeWaitTime = 30 * time.Second
)
var (
	ErrStreamExists = errors.New("stream already exists")
	ErrNoSuchStream = errors.New("stream does not exist")
	ErrNoSuchPartition = errors.New("stream partition does not exist")
	ErrStreamDeleted = errors.New("stream has been deleted")
	ErrPartitionPaused = errors.New("stream partition has been paused")
	ErrReadonlyPartition = errors.New("readonly partition")
)
type Handler func(msg *Message, err error)
type StreamOptions struct {
	Group string
	ReplicationFactor int32
	Partitions int32
}
type StreamOption func(*StreamOptions) error
func Group(group string) StreamOption {
	return func(o *StreamOptions) error {
		o.Group = group
		return nil
	}
}
func ReplicationFactor(replicationFactor int32) StreamOption {
	return func(o *StreamOptions) error {
		o.ReplicationFactor = replicationFactor
		return nil
	}
}
func MaxReplication() StreamOption {
	return func(o *StreamOptions) error {
		o.ReplicationFactor = MaxReplicationFactor
		return nil
	}
}
func Partitions(partitions int32) StreamOption {
	return func(o *StreamOptions) error {
		if partitions < 0 {
			return fmt.Errorf("invalid number of partitions: %d", partitions)
		}
		o.Partitions = partitions
		return nil
	}
}
type Client interface {
	Close() error
	CreateStream(ctx context.Context, subject, name string, opts ...StreamOption) error
	DeleteStream(ctx context.Context, name string) error
	PauseStream(ctx context.Context, name string, opts ...PauseOption) error
	SetStreamReadonly(ctx context.Context, name string, opts ...ReadonlyOption) error
	Subscribe(ctx context.Context, stream string, handler Handler, opts ...SubscriptionOption) error
	Publish(ctx context.Context, stream string, value []byte, opts ...MessageOption) (*Ack, error)
	PublishToSubject(ctx context.Context, subject string, value []byte, opts ...MessageOption) (*Ack, error)
	FetchMetadata(ctx context.Context) (*Metadata, error)
}
type client struct {
	mu        sync.RWMutex
	apiClient proto.APIClient
	conn      *grpc.ClientConn
	metadata  *metadataCache
	pools     map[string]*connPool
	opts      ClientOptions
	dialOpts  []grpc.DialOption
	closed    bool
}
type ClientOptions struct {
	Brokers []string
	MaxConnsPerBroker int
	KeepAliveTime time.Duration
	TLSCert string
	TLSConfig *tls.Config
	ResubscribeWaitTime time.Duration
}
func (o ClientOptions) Connect() (Client, error) {
	if len(o.Brokers) == 0 {
		return nil, errors.New("no addresses provided")
	}
	var (
		conn *grpc.ClientConn
		err  error
		opts = []grpc.DialOption{}
	)
	if o.TLSConfig != nil {
		creds := credentials.NewTLS(o.TLSConfig)
		opts = append(opts, grpc.WithTransportCredentials(creds))
	} else if o.TLSCert != "" {
		creds, err := credentials.NewClientTLSFromFile(o.TLSCert, "")
		if err != nil {
			return nil, fmt.Errorf("could not load tls cert: %s", err)
		}
		opts = append(opts, grpc.WithTransportCredentials(creds))
	} else {
		opts = append(opts, grpc.WithInsecure())
	}
	perm := rand.Perm(len(o.Brokers))
	for _, i := range perm {
		addr := o.Brokers[i]
		conn, err = grpc.Dial(addr, opts...)
		if err == nil {
			break
		}
	}
	if conn == nil {
		return nil, err
	}
	c := &client{
		conn:      conn,
		apiClient: proto.NewAPIClient(conn),
		pools:     make(map[string]*connPool),
		opts:      o,
		dialOpts:  opts,
	}
	c.metadata = newMetadataCache(o.Brokers, c.doResilientRPC)
	if _, err := c.metadata.update(context.Background()); err != nil {
		return nil, err
	}
	return c, nil
}
func DefaultClientOptions() ClientOptions {
	return ClientOptions{
		MaxConnsPerBroker:   defaultMaxConnsPerBroker,
		KeepAliveTime:       defaultKeepAliveTime,
		ResubscribeWaitTime: defaultResubscribeWaitTime,
	}
}
type ClientOption func(*ClientOptions) error
func MaxConnsPerBroker(max int) ClientOption {
	return func(o *ClientOptions) error {
		o.MaxConnsPerBroker = max
		return nil
	}
}