func MinISR(minISR int) StreamOption {
	return func(o *StreamOptions) error {
		o.MinISR = &minISR
		return nil
	}
}
MinISR overrides clustering.min.insync.replicas for the given stream. This
controls the minimum number of replicas that must acknowledge a stream write
before it can be committed. If this is not set, it uses the server default
value.
package liftbridge
import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"
	"github.com/nats-io/nuid"
	"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)
}
type StopPosition int32
func (s StopPosition) toProto() proto.StopPosition {
	return proto.StopPosition(s)
}
const (
	defaultMaxConnsPerBroker   = 2
	defaultKeepAliveTime       = 30 * time.Second
	defaultResubscribeWaitTime = 30 * time.Second
	defaultAckWaitTime         = 5 * time.Second
	cursorsStream = "__cursors"
)
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")
	ErrAckTimeout = errors.New("publish ack timeout")
	ErrReadonlyPartition = errors.New("readonly partition")
)
type Handler func(msg *Message, err error)
type AckHandler func(ack *Ack, err error)
type ackContext struct {
	handler AckHandler
	timer   *time.Timer
}
type StreamOptions struct {
	Group string
	ReplicationFactor int32
	Partitions int32
	RetentionMaxBytes *int64
	RetentionMaxMessages *int64
	RetentionMaxAge *time.Duration
	CleanerInterval *time.Duration
	SegmentMaxBytes *int64
	SegmentMaxAge *time.Duration
	CompactMaxGoroutines *int32
	CompactEnabled *bool
	AutoPauseTime *time.Duration
	AutoPauseDisableIfSubscribers *bool
	MinISR *int
	OptimisticConcurrencyControl *bool
	Encryption *bool
}
func (s *StreamOptions) newRequest(subject, name string) *proto.CreateStreamRequest {
	req := &proto.CreateStreamRequest{
		Subject: subject,
		Name:    name,
	}
	req.ReplicationFactor = s.ReplicationFactor
	req.Group = s.Group
	req.Partitions = s.Partitions
	if s.RetentionMaxAge != nil {
		req.RetentionMaxAge = &proto.NullableInt64{Value: s.RetentionMaxAge.Milliseconds()}
	}
	if s.RetentionMaxBytes != nil {
		req.RetentionMaxBytes = &proto.NullableInt64{Value: *s.RetentionMaxBytes}
	}
	if s.RetentionMaxMessages != nil {
		req.RetentionMaxMessages = &proto.NullableInt64{Value: *s.RetentionMaxMessages}
	}
	if s.CleanerInterval != nil {
		req.CleanerInterval = &proto.NullableInt64{Value: s.CleanerInterval.Milliseconds()}
	}
	if s.SegmentMaxBytes != nil {
		req.SegmentMaxBytes = &proto.NullableInt64{Value: *s.SegmentMaxBytes}
	}
	if s.SegmentMaxAge != nil {
		req.SegmentMaxAge = &proto.NullableInt64{Value: s.SegmentMaxAge.Milliseconds()}
	}
	if s.CompactMaxGoroutines != nil {
		req.CompactMaxGoroutines = &proto.NullableInt32{Value: *s.CompactMaxGoroutines}
	}
	if s.CompactEnabled != nil {
		req.CompactEnabled = &proto.NullableBool{Value: *s.CompactEnabled}
	}
	if s.AutoPauseTime != nil {
		req.AutoPauseTime = &proto.NullableInt64{Value: s.AutoPauseTime.Milliseconds()}
	}
	if s.AutoPauseDisableIfSubscribers != nil {
		req.AutoPauseDisableIfSubscribers = &proto.NullableBool{Value: *s.AutoPauseDisableIfSubscribers}
	}
	if s.MinISR != nil {
		req.MinIsr = &proto.NullableInt32{Value: int32(*s.MinISR)}
	}
	if s.OptimisticConcurrencyControl != nil {
		req.OptimisticConcurrencyControl = &proto.NullableBool{Value: *s.OptimisticConcurrencyControl}
	}
	if s.Encryption != nil {
		req.Encryption = &proto.NullableBool{Value: *s.Encryption}
	}
	return req
}
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
	}
}
func RetentionMaxBytes(val int64) StreamOption {
	return func(o *StreamOptions) error {
		o.RetentionMaxBytes = &val
		return nil
	}
}
func RetentionMaxMessages(val int64) StreamOption {
	return func(o *StreamOptions) error {
		o.RetentionMaxMessages = &val
		return nil
	}
}
func RetentionMaxAge(val time.Duration) StreamOption {
	return func(o *StreamOptions) error {
		o.RetentionMaxAge = &val
		return nil
	}
}
func CleanerInterval(val time.Duration) StreamOption {
	return func(o *StreamOptions) error {
		o.CleanerInterval = &val
		return nil
	}
}
func SegmentMaxBytes(val int64) StreamOption {
	return func(o *StreamOptions) error {
		o.SegmentMaxBytes = &val
		return nil
	}
}
func SegmentMaxAge(val time.Duration) StreamOption {
	return func(o *StreamOptions) error {
		o.SegmentMaxAge = &val
		return nil
	}
}
func CompactMaxGoroutines(val int32) StreamOption {
	return func(o *StreamOptions) error {
		o.CompactMaxGoroutines = &val
		return nil
	}
}
func CompactEnabled(val bool) StreamOption {
	return func(o *StreamOptions) error {
		o.CompactEnabled = &val
		return nil
	}
}
func AutoPauseTime(val time.Duration) StreamOption {
	return func(o *StreamOptions) error {
		o.AutoPauseTime = &val
		return nil
	}
}
func AutoPauseDisableIfSubscribers(val bool) StreamOption {
	return func(o *StreamOptions) error {
		o.AutoPauseDisableIfSubscribers = &val
		return nil
	}
}