func StartAtTimeDelta(ago time.Duration) SubscriptionOption {
	return func(o *SubscriptionOptions) error {
		o.StartPosition = StartPosition(proto.StartPosition_TIMESTAMP)
		o.StartTimestamp = time.Now().Add(-ago)
		return nil
	}
}
StartAtTimeDelta sets the desired timestamp to begin consuming from in the
stream using a time delta in the past.
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
	}
}
func KeepAliveTime(keepAlive time.Duration) ClientOption {
	return func(o *ClientOptions) error {
		o.KeepAliveTime = keepAlive
		return nil
	}
}
func TLSCert(cert string) ClientOption {
	return func(o *ClientOptions) error {
		o.TLSCert = cert
		return nil
	}
}
func TLSConfig(config *tls.Config) ClientOption {
	return func(o *ClientOptions) error {
		o.TLSConfig = config
		return nil
	}
}
func ResubscribeWaitTime(wait time.Duration) ClientOption {
	return func(o *ClientOptions) error {
		o.ResubscribeWaitTime = wait
		return nil
	}
}
func Connect(addrs []string, options ...ClientOption) (Client, error) {
	opts := DefaultClientOptions()
	opts.Brokers = addrs
	for _, opt := range options {
		if err := opt(&opts); err != nil {
			return nil, err
		}
	}
	return opts.Connect()
}
func (c *client) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil
	}
	for _, pool := range c.pools {
		if err := pool.close(); err != nil {
			return err
		}
	}
	if err := c.conn.Close(); err != nil {
		return err
	}
	c.closed = true
	return nil
}
func (c *client) CreateStream(ctx context.Context, subject, name string, options ...StreamOption) error {
	opts := &StreamOptions{}
	for _, opt := range options {
		if err := opt(opts); err != nil {
			return err
		}
	}
	req := &proto.CreateStreamRequest{
		Subject:           subject,
		Name:              name,
		ReplicationFactor: opts.ReplicationFactor,
		Group:             opts.Group,
		Partitions:        opts.Partitions,
	}
	err := c.doResilientRPC(func(client proto.APIClient) error {
		_, err := client.CreateStream(ctx, req)
		return err
	})
	if status.Code(err) == codes.AlreadyExists {
		return ErrStreamExists
	}
	return err
}
func (c *client) DeleteStream(ctx context.Context, name string) error {
	req := &proto.DeleteStreamRequest{Name: name}
	err := c.doResilientRPC(func(client proto.APIClient) error {
		_, err := client.DeleteStream(ctx, req)
		return err
	})
	if status.Code(err) == codes.NotFound {
		return ErrNoSuchStream
	}
	return err
}
type PauseOptions struct {
	Partitions []int32
	ResumeAll bool
}
type PauseOption func(*PauseOptions) error
func PausePartitions(partitions ...int32) PauseOption {
	return func(o *PauseOptions) error {
		o.Partitions = partitions
		return nil
	}
}
func ResumeAll() PauseOption {
	return func(o *PauseOptions) error {
		o.ResumeAll = true
		return nil
	}
}
func (c *client) PauseStream(ctx context.Context, name string, options ...PauseOption) error {
	opts := &PauseOptions{}
	for _, opt := range options {
		if err := opt(opts); err != nil {
			return err
		}
	}
	req := &proto.PauseStreamRequest{
		Name:       name,
		Partitions: opts.Partitions,
		ResumeAll:  opts.ResumeAll,
	}
	err := c.doResilientRPC(func(client proto.APIClient) error {
		_, err := client.PauseStream(ctx, req)
		return err
	})
	if status.Code(err) == codes.NotFound {
		return ErrNoSuchPartition
	}
	return err
}
type ReadonlyOptions struct {
	Partitions []int32
	Readwrite bool
}
type ReadonlyOption func(*ReadonlyOptions) error
func ReadonlyPartitions(partitions ...int32) ReadonlyOption {
	return func(o *ReadonlyOptions) error {
		o.Partitions = partitions
		return nil
	}
}
func Readonly(readonly bool) ReadonlyOption {
	return func(o *ReadonlyOptions) error {
		o.Readwrite = !readonly
		return nil
	}
}
func (c *client) SetStreamReadonly(ctx context.Context, name string, options ...ReadonlyOption) error {
	opts := &ReadonlyOptions{}
	for _, opt := range options {
		if err := opt(opts); err != nil {
			return err
		}
	}
	req := &proto.SetStreamReadonlyRequest{
		Name:       name,
		Partitions: opts.Partitions,
		Readonly:   !opts.Readwrite,
	}
	err := c.doResilientRPC(func(client proto.APIClient) error {
		_, err := client.SetStreamReadonly(ctx, req)
		return err
	})
	if status.Code(err) == codes.NotFound {
		return ErrNoSuchPartition
	}
	return err
}
type SubscriptionOptions struct {
	StartPosition StartPosition
	StartOffset int64
	StartTimestamp time.Time
	Partition int32
	ReadISRReplica bool
	Resume bool
}
type SubscriptionOption func(*SubscriptionOptions) error
func StartAtOffset(offset int64) SubscriptionOption {
	return func(o *SubscriptionOptions) error {
		o.StartPosition = StartPosition(proto.StartPosition_OFFSET)
		o.StartOffset = offset
		return nil
	}
}
func StartAtTime(start time.Time) SubscriptionOption {
	return func(o *SubscriptionOptions) error {
		o.StartPosition = StartPosition(proto.StartPosition_TIMESTAMP)
		o.StartTimestamp = start
		return nil
	}
}