natty

type AsLeaderConfig struct {
	// Looper is the loop construct that will be used to execute Func (required)
	Looper director.Looper

	// Bucket specifies what K/V bucket will be used for leader election (required)
	Bucket string

	// Key specifies the keyname that the leader election will occur on (required)
	Key string

	// NodeName is the name used for this node; should be unique in cluster (required)
	NodeName string

	// Description will set the bucket description (optional)
	Description string

	// ElectionLooper allows you to override the used election looper (optional)
	ElectionLooper director.Looper

	// BucketTTL specifies the TTL policy the bucket should use (optional)
	BucketTTL time.Duration

	// ReplicaCount specifies the number of replicas the bucket should use (optional, default 1)
	ReplicaCount int
}

type Config struct {
	// NatsURL defines the NATS urls the library will attempt to connect to. Iff
	// first URL fails, we will try to connect to the next one. Only fail if all
	// URLs fail.
	NatsURL []string

	// MaxMsgs defines the maximum number of messages a stream will contain.
	MaxMsgs int64

	// FetchSize defines the number of messages to fetch from the stream during
	// a single Fetch() call.
	FetchSize int

	// FetchTimeout defines how long a Fetch() call will wait to attempt to reach
	// defined FetchSize before continuing.
	FetchTimeout time.Duration

	// DeliverPolicy defines the policy the library will use to deliver messages.
	// Default: DeliverLastPolicy which will deliver from the last message that
	// the consumer has seen.
	DeliverPolicy nats.DeliverPolicy

	// Logger allows you to inject a logger into the library. Optional.
	Logger Logger

	// Whether to use TLS
	UseTLS bool

	// TLS CA certificate file
	TLSCACertFile string

	// TLS client certificate file
	TLSClientCertFile string

	// TLS client key file
	TLSClientKeyFile string

	// Do not perform server certificate checks
	TLSSkipVerify bool

	// PublishBatchSize is how many messages to async publish at once
	// Default: 256
	PublishBatchSize int

	// ServiceShutdownContext is used by main() to shutdown services before application termination
	ServiceShutdownContext context.Context

	// MainShutdownFunc is triggered by watchForShutdown() after all publisher queues are exhausted
	// and is used to trigger shutdown of APIs and then main()
	MainShutdownFunc context.CancelFunc

	// WorkerIdleTimeout determines how long to keep a publish worker alive if no activity
	WorkerIdleTimeout time.Duration

	// PublishTimeout is how long to wait for a batch of async publish calls to be ACK'd
	PublishTimeout time.Duration

	// PublishErrorCh will receive any
	PublishErrorCh chan *PublishError
}

type ConsumerConfig struct {
	// Subject is the subject to consume off of a stream
	Subject string

	// StreamName is the name of JS stream to consume from.
	// This should first be created with CreateStream()
	StreamName string

	// ConsumerName is the consumer that was made with CreateConsumer()
	ConsumerName string

	// Looper is optional, if none is provided, one will be created
	Looper director.Looper

	// ErrorCh is used to retrieve any errors returned during asynchronous publishing
	// If nil, errors will only be logged
	ErrorCh chan error
}

type INatty interface {
	// Consume subscribes to given subject and executes callback every time a
	// message is received. Consumed messages must be explicitly ACK'd or NAK'd.
	//
	// This is a blocking call; cancellation should be performed via the context.
	Consume(ctx context.Context, cfg *ConsumerConfig, cb func(ctx context.Context, msg *nats.Msg) error) error

	// Publish publishes a single message with the given subject; this method
	// will perform automatic batching as configured during `natty.New(..)`
	Publish(ctx context.Context, subject string, data []byte)

	// DeletePublisher shuts down a publisher and deletes it from the internal publisherMap
	DeletePublisher(ctx context.Context, id string) bool

	// CreateStream creates a new stream if it does not exist
	CreateStream(ctx context.Context, name string, subjects []string) error

	// DeleteStream deletes an existing stream
	DeleteStream(ctx context.Context, name string) error

	// CreateConsumer creates a new consumer if it does not exist
	CreateConsumer(ctx context.Context, streamName, consumerName string, filterSubject ...string) error

	// DeleteConsumer deletes an existing consumer
	DeleteConsumer(ctx context.Context, consumerName, streamName string) error

	// Get will fetch the value for a given bucket and key. Will NOT auto-create
	// bucket if it does not exist.
	Get(ctx context.Context, bucket string, key string) ([]byte, error)

	// Create will attempt to create a key in KV. It will return an error if
	// the key already exists. Will auto-create the bucket if it does not
	// already exist.
	Create(ctx context.Context, bucket string, key string, data []byte, keyTTL ...time.Duration) error

	// Put will put a new value for a given bucket and key. Will auto-create
	// the bucket if it does not already exist.
	Put(ctx context.Context, bucket string, key string, data []byte, ttl ...time.Duration) error

	// Delete will delete a key from a given bucket. Will no-op if the bucket
	// or key does not exist.
	Delete(ctx context.Context, bucket string, key string) error

	// CreateBucket will attempt to create a new bucket. Will return an error if
	// bucket already exists.
	CreateBucket(ctx context.Context, bucket string, ttl time.Duration, replicas int, description ...string) error

	// DeleteBucket will delete the specified bucket
	DeleteBucket(ctx context.Context, bucket string) error

	// WatchBucket returns an instance of nats.KeyWatcher for the given bucket
	WatchBucket(ctx context.Context, bucket string) (nats.KeyWatcher, error)

	// WatchKey returns an instance of nats.KeyWatcher for the given bucket and key
	WatchKey(ctx context.Context, bucket, key string) (nats.KeyWatcher, error)

	// Keys will return all of the keys in a bucket (empty slice if none found)
	Keys(ctx context.Context, bucket string) ([]string, error)

	// Refresh will attempt to perform a "safe" refresh of a key that has a TTL.
	// Natty will first attempt to fetch the key so that it can get its revision
	// and then perform an Update() referencing the revision. If the revision
	// does not match, it will return an error.
	Refresh(ctx context.Context, bucket, key string) error

	// Status queries the status of the KV bucket
	Status(ctx context.Context, bucket string) (nats.KeyValueStatus, error)

	// AsLeader enables simple leader election by using NATS k/v functionality.
	//
	// AsLeader will execute opts.Func if and only if the node executing AsLeader
	// acquires leader role. It will continue executing opts.Func until it loses
	// leadership and another node becomes leader.
	AsLeader(ctx context.Context, opts AsLeaderConfig, f func() error) error

	// HaveLeader returns bool indicating whether node-name in given cfg is the
	// leader for the cfg.Bucket and cfg.Key
	HaveLeader(ctx context.Context, nodeName, bucketName, keyName string) bool

	// GetLeader returns the current leader for a given bucket and key
	GetLeader(ctx context.Context, bucketName, keyName string) (string, error)
}

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

type Logger interface {
	// Debug sends out a debug message with the given arguments to the logger.
	Debug(args ...interface{})
	// Debugf formats a debug message using the given arguments and sends it to the logger.
	Debugf(format string, args ...interface{})
	// Info sends out an informational message with the given arguments to the logger.
	Info(args ...interface{})
	// Infof formats an informational message using the given arguments and sends it to the logger.
	Infof(format string, args ...interface{})
	// Warn sends out a warning message with the given arguments to the logger.
	Warn(args ...interface{})
	// Warnf formats a warning message using the given arguments and sends it to the logger.
	Warnf(format string, args ...interface{})
	// Error sends out an error message with the given arguments to the logger.
	Error(args ...interface{})
	// Errorf formats an error message using the given arguments and sends it to the logger.
	Errorf(format string, args ...interface{})
}

type Mode int

type Natty struct {
	*Config
	// contains filtered or unexported fields
}

type NoOpLogger struct {
}

type PublishError struct {
	Subject string
	Message error
}

type Publisher struct {
	Subject     string
	QueueMutex  *sync.RWMutex
	Queue       []*message
	Natty       *Natty
	IdleTimeout time.Duration

	// ErrorCh is optional. It will receive async publish errors if specified
	// Otherwise errors will only be logged
	ErrorCh chan *PublishError

	// PublisherContext is used to close a specific publisher
	PublisherContext context.Context

	// PublisherCancel is used to cancel a specific publisher's context
	PublisherCancel context.CancelFunc

	// ServiceShutdownContext is used by main() to shutdown services before application termination
	ServiceShutdownContext context.Context
	// contains filtered or unexported fields
}