liftbridge

type Ack struct {
	// contains filtered or unexported fields
}

type AckHandler func(ack *Ack, err error)

type AckPolicy int32

type AutoOffset int

type BrokerInfo struct {
	// contains filtered or unexported fields
}

type Client interface {
	// Close the client connection.
	Close() error

	// CreateStream creates a new stream attached to a NATS subject. Subject is
	// the NATS subject the stream is attached to, and name is the stream
	// identifier, unique per subject. It returns ErrStreamExists if a stream
	// with the given subject and name already exists.
	CreateStream(ctx context.Context, subject, name string, opts ...StreamOption) error

	// DeleteStream deletes a stream and all of its partitions. Name is the
	// stream identifier, globally unique.
	DeleteStream(ctx context.Context, name string) error

	// PauseStream pauses a stream and some or all of its partitions. Name is
	// the stream identifier, globally unique. It returns an ErrNoSuchPartition
	// if the given stream or partition does not exist. By default, this will
	// pause all partitions. A partition is resumed when it is published to via
	// the Liftbridge Publish API or ResumeAll is enabled and another partition
	// in the stream is published to.
	PauseStream(ctx context.Context, name string, opts ...PauseOption) error

	// SetStreamReadonly sets the readonly flag on a stream and some or all of
	// its partitions. Name is the stream identifier, globally unique. It
	// returns an ErrNoSuchPartition if the given stream or partition does not
	// exist. By default, this will set the readonly flag on all partitions.
	// Subscribers to a readonly partition will see their subscription ended
	// with a ErrReadonlyPartition error once all messages currently in the
	// partition have been read.
	SetStreamReadonly(ctx context.Context, name string, opts ...ReadonlyOption) error

	// Subscribe creates an ephemeral subscription for the given stream. It
	// begins receiving messages starting at the configured position and waits
	// for new messages when it reaches the end of the stream. The default
	// start position is the end of the stream.
	// ErrNoSuchPartition is returned if the given stream or partition does not
	// exist.
	// ErrReadonlyPartition is return to subscribers when all messages have been
	// read from a read only stream, or when the configured stop position is
	// reached.
	// Use a cancelable Context to close a subscription.
	Subscribe(ctx context.Context, stream string, handler Handler, opts ...SubscriptionOption) error

	// Publish publishes a new message to the Liftbridge stream. The partition
	// that gets published to is determined by the provided partition or
	// Partitioner passed through MessageOptions, if any. If a partition or
	// Partitioner is not provided, this defaults to the base partition. This
	// partition determines the underlying NATS subject that gets published to.
	// To publish directly to a specific NATS subject, use the low-level
	// PublishToSubject API.
	//
	// If the AckPolicy is not NONE, this will synchronously block until the
	// ack is received. If the ack is not received in time, ErrAckTimeout is
	// returned. If AckPolicy is NONE, this returns nil on success.
	Publish(ctx context.Context, stream string, value []byte, opts ...MessageOption) (*Ack, error)

	// PublishAsync publishes a new message to the Liftbridge stream and
	// asynchronously processes the ack or error for the message.
	PublishAsync(ctx context.Context, stream string, value []byte, ackHandler AckHandler, opts ...MessageOption) error

	// PublishToSubject publishes a new message to the NATS subject. Note that
	// because this publishes directly to a subject, there may be multiple (or
	// no) streams that receive the message. As a result, MessageOptions
	// related to partitioning will be ignored. To publish at the
	// stream/partition level, use the high-level Publish API.
	//
	// If the AckPolicy is not NONE and a deadline is provided, this will
	// synchronously block until the first ack is received. If an ack is not
	// received in time, ErrAckTimeout is returned. If an AckPolicy and
	// deadline are configured, this returns the first Ack on success,
	// otherwise it returns nil.
	PublishToSubject(ctx context.Context, subject string, value []byte, opts ...MessageOption) (*Ack, error)

	// FetchMetadata returns cluster metadata including broker and stream
	// information.
	FetchMetadata(ctx context.Context, opts ...MetadataOption) (*Metadata, error)

	// SetCursor persists a cursor position for a particular stream partition.
	// This can be used to checkpoint a consumer's position in a stream to
	// resume processing later.
	SetCursor(ctx context.Context, id, stream string, partition int32, offset int64) error

	// FetchCursor retrieves a cursor position for a particular stream
	// partition. It returns -1 if the cursor does not exist.
	FetchCursor(ctx context.Context, id, stream string, partition int32) (int64, error)

	// FetchPartitionMetadata retrieves the metadata of a particular partition
	FetchPartitionMetadata(ctx context.Context, stream string, partition int32) (*PartitionInfo, error)

	// CreateConsumer creates a consumer that is part of a consumer group which
	// consumes a set of streams. Liftbridge handles assigning partitions to
	// the group members and tracking the group's position in the streams.
	CreateConsumer(groupID string, opts ...ConsumerOption) (*Consumer, error)
}

type ClientOption func(*ClientOptions) error

type ClientOptions struct {
	// Brokers it the set of hosts the client will use when attempting to
	// connect.
	Brokers []string

	// Deprecated: MaxConnsPerBroker is no longer used since the client now
	// maintains one connection per broker.
	MaxConnsPerBroker int

	// Deprecated: KeepAliveTime is no longer used since the client now
	// maintains one connection per broker.
	KeepAliveTime time.Duration

	// TLSCert is the TLS certificate file to use. The client does not use a
	// TLS connection if this is not set.
	TLSCert string

	// TLSConfig is the TLS configuration to use. The client does not use a
	// TLS connection if this is not set. Overrides TLSCert if set.
	TLSConfig *tls.Config

	// ResubscribeWaitTime is the amount of time to attempt to re-establish a
	// stream subscription after being disconnected. For example, if the server
	// serving a subscription dies and the stream is replicated, the client
	// will attempt to re-establish the subscription once the stream leader has
	// failed over. This failover can take several moments, so this option
	// gives the client time to retry. The default is 30 seconds.
	ResubscribeWaitTime time.Duration

	// AckWaitTime is the default amount of time to wait for an ack to be
	// received for a published message before ErrAckTimeout is returned. This
	// can be overridden on individual requests by setting a timeout on the
	// Context. This defaults to 5 seconds if not set.
	AckWaitTime time.Duration

	// ReadBufferSize configures the size of the read buffer for connections.
	// -1 will use the GRPC defaults. See
	// https://godoc.org/google.golang.org/grpc#WithReadBufferSize.
	ReadBufferSize int

	// WriteBufferSize configures the size of the write buffer for connections.
	// -1 will use the GRPC defaults. See
	// https://godoc.org/google.golang.org/grpc#WithWriteBufferSize.
	WriteBufferSize int

	// Select the server to connecto to based on specific criteria
	ServerSelection SelectionCriteria
}

type Consumer struct {
	// contains filtered or unexported fields
}

type ConsumerError struct {
	// contains filtered or unexported fields
}

type ConsumerGroupInfo struct {
	// contains filtered or unexported fields
}

type ConsumerOption func(*ConsumerOptions) error

type ConsumerOptions struct {
	// ConsumerID uniquely identifies a logical consumer. This ID should not be
	// reused across consumers. If a ConsumerID is not supplied, a random one
	// will be generated.
	ConsumerID string

	// AutoCheckpointInterval determines the frequency the consumer's positions
	// are committed to Liftbridge. A value of 0 disables auto checkpointing.
	// The default value is 5 seconds if not set.
	AutoCheckpointInterval time.Duration

	// AutoOffsetDefault determines the behavior for where a consumer should
	// start consuming a stream partition when the consumer group does not have
	// a committed offset, e.g. because the group was just created. If not set,
	// defaults to AutoOffsetNewOnly.
	AutoOffset AutoOffset

	// FetchAssignmentsInterval is a function which returns the frequency to
	// fetch partition assignments from the consumer group coordinator. This
	// also acts as a health check to keep the consumer active in the group.
	// Increasing this too much may cause the group coordinator to think the
	// consumer has failed and remove it from the group. The function argument
	// is the timeout duration configured on the server. If not set, this will
	// default to 0.4 * timeout.
	FetchAssignmentsInterval func(timeout time.Duration) time.Duration
}

type Handler func(msg *Message, err error)

type Message struct {
	// contains filtered or unexported fields
}

type MessageOption func(*MessageOptions)

type MessageOptions struct {
	// Key to set on the Message. If Liftbridge has stream compaction enabled,
	// the stream will retain only the last value for each key.
	Key []byte

	// AckInbox sets the NATS subject Liftbridge should publish the Message ack
	// to. If this is not set, the server will generate a random inbox.
	AckInbox string

	// CorrelationID sets the identifier used to correlate an ack with the
	// published Message. If this is not set, the ack will not have a
	// correlation id.
	CorrelationID string

	// AckPolicy controls the behavior of Message acks sent by the server. By
	// default, Liftbridge will send an ack when the stream leader has written
	// the Message to its write-ahead log.
	AckPolicy AckPolicy

	// Headers are key-value pairs to set on the Message.
	Headers map[string][]byte

	// Partitioner specifies the strategy for mapping a Message to a stream
	// partition.
	Partitioner Partitioner

	// Partition specifies the stream partition to publish the Message to. If
	// this is set, any Partitioner will not be used. This is a pointer to
	// allow distinguishing between unset and 0.
	Partition *int32

	// ExpectedOffset set the value of the expected offset after publishing the message.
	// This is required in case Optimistic Concurrency Control is activted
	// by default, this value should be set to -1 to indicate next offset.
	ExpectedOffset int64
}

type Metadata struct {
	// contains filtered or unexported fields
}

type MetadataOption func(*MetadataOptions) error

type MetadataOptions struct {
	// ConsumerGroups determines the consumer groups to fetch metadata for. If
	// not set, no consumer group metadata will be fetched.
	ConsumerGroups []string

	// Streams determines the streams to fetch metadata for. If not set,
	// metadata for all streams will be fetched.
	Streams []string
	// contains filtered or unexported fields
}

type PartitionEventTimestamps struct {
	// contains filtered or unexported fields
}

type PartitionInfo struct {
	// contains filtered or unexported fields
}

type Partitioner interface {
	// Partition computes the partition number for a given message.
	Partition(stream string, key, value []byte, metadata *Metadata) int32
}

type PauseOption func(*PauseOptions) error

type PauseOptions struct {
	// Partitions sets the list of partitions to pause or all of them if
	// nil/empty.
	Partitions []int32

	// ResumeAll will resume all partitions in the stream if one of them is
	// published to instead of resuming only that partition.
	ResumeAll bool
}

type ReadonlyOption func(*ReadonlyOptions) error

type ReadonlyOptions struct {
	// Partitions sets the list of partitions on which to set the readonly flag
	// or all of them if nil/empty.
	Partitions []int32

	// Readwrite defines if the partitions should be set to readonly (false) or
	// to readwrite (true). This field is called readwrite and not readonly so
	// that the default value corresponds to "enable readonly".
	Readwrite bool
}

type SelectionCriteria int

type StartPosition int32

type StopPosition int32

type StreamInfo struct {
	// contains filtered or unexported fields
}

type StreamOption func(*StreamOptions) error

type StreamOptions struct {
	// Group is the name of a load-balance group. When there are multiple
	// streams in the same group, messages will be balanced among them.
	Group string

	// ReplicationFactor controls the number of servers to replicate a stream
	// to. E.g. a value of 1 would mean only 1 server would have the data, and
	// a value of 3 would be 3 servers would have it. If this is not set, it
	// defaults to 1. A value of -1 will signal to the server to set the
	// replication factor equal to the current number of servers in the
	// cluster.
	ReplicationFactor int32

	// Partitions determines how many partitions to create for a stream. If 0,
	// this will behave as a stream with a single partition. If this is not
	// set, it defaults to 1.
	Partitions int32

	// The maximum size a stream's log can grow to, in bytes, before we will
	// discard old log segments to free up space. A value of 0 indicates no
	// limit. If this is not set, it uses the server default value.
	RetentionMaxBytes *int64

	// The maximum size a stream's log can grow to, in number of messages,
	// before we will discard old log segments to free up space. A value of 0
	// indicates no limit. If this is not set, it uses the server default
	// value.
	RetentionMaxMessages *int64

	// The TTL for stream log segment files, after which they are deleted. A
	// value of 0 indicates no TTL. If this is not set, it uses the server
	// default value.
	RetentionMaxAge *time.Duration

	// The frequency to check if a new stream log segment file should be rolled
	// and whether any segments are eligible for deletion based on the
	// retention policy or compaction if enabled. If this is not set, it uses
	// the server default value.
	CleanerInterval *time.Duration

	// The maximum size of a single stream log segment file in bytes. Retention
	// is always done a file at a time, so a larger segment size means fewer
	// files but less granular control over retention. If this is not set, it
	// uses the server default value.
	SegmentMaxBytes *int64

	// The maximum time before a new stream log segment is rolled out. A value
	// of 0 means new segments will only be rolled when segment.max.bytes is
	// reached. Retention is always done a file at a time, so a larger value
	// means fewer files but less granular control over retention. If this is
	// not set, it uses the server default value.
	SegmentMaxAge *time.Duration

	// The maximum number of concurrent goroutines to use for compaction on a
	// stream log (only applicable if compact.enabled is true). If this is not
	// set, it uses the server default value.
	CompactMaxGoroutines *int32

	// CompactEnabled controls the activation of stream log compaction. If this
	// is not set, it uses the server default value.
	CompactEnabled *bool

	// The amount of time a stream partition can go idle before it is
	// automatically paused. If this is not set, it uses the server default
	// value.
	AutoPauseTime *time.Duration

	// Disables automatic partition pausing when there are subscribers. If this
	// is not set, it uses the server default value.
	AutoPauseDisableIfSubscribers *bool

	// 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.
	MinISR *int

	// OptimisticConcurrencyControl controls the activation of optimistic concurrency control
	// of the stream
	OptimisticConcurrencyControl *bool

	// Encryption controls the activation of encryption data-at-rest on server side
	Encryption *bool
}

type SubscriptionOption func(*SubscriptionOptions) error

type SubscriptionOptions struct {
	// StartPosition controls where to begin consuming from in the stream.
	StartPosition StartPosition

	// StartOffset sets the stream offset to begin consuming from.
	StartOffset int64

	// StartTimestamp sets the stream start position to the given timestamp.
	StartTimestamp time.Time

	// StopPosition controls where to stop consuming in the stream.
	StopPosition StopPosition

	// StopOffset sets the stream offset to stop consuming at.
	StopOffset int64

	// StopTimestamp sets the stream stop position to the given timestamp.
	StopTimestamp time.Time

	// Partition sets the stream partition to consume.
	Partition int32

	// ReadISRReplica sets client's ability to subscribe from a random ISR.
	ReadISRReplica bool

	// Resume controls if a paused partition can be resumed before
	// subscription.
	Resume bool
	// contains filtered or unexported fields
}