Versions in this module Expand all Collapse all v1 v1.38.0 Nov 16, 2024 v1.37.0 Nov 16, 2024 Changes in this version + const AllKeys + const DefaultAPIPrefix + const DefaultExpires + const DefaultMaxMessages + const DefaultPubRetryAttempts + const DefaultPubRetryWait + const ExpectedLastMsgIDHeader + const ExpectedLastSeqHeader + const ExpectedLastSubjSeqHeader + const ExpectedStreamHeader + const KeyValueMaxHistory + const LastSequenceHeader + const MsgIDHeader + const MsgRollup + const MsgRollupAll + const MsgRollupSubject + const SequenceHeader + const StreamHeader + const SubjectHeader + const TimeStampHeaer + func DecodeObjectDigest(data string) ([]byte, error) + func GetObjectDigestValue(data hash.Hash) string + type APIError struct + Code int + Description string + ErrorCode ErrorCode + func (e *APIError) APIError() *APIError + func (e *APIError) Error() string + func (e *APIError) Is(err error) bool + type APIStats struct + Errors uint64 + Total uint64 + type AccountInfo struct + API APIStats + Domain string + Tiers map[string]Tier + type AccountLimits struct + MaxConsumers int + MaxMemory int64 + MaxStore int64 + MaxStreams int + type AckPolicy int + const AckAllPolicy + const AckExplicitPolicy + const AckNonePolicy + func (p *AckPolicy) UnmarshalJSON(data []byte) error + func (p AckPolicy) MarshalJSON() ([]byte, error) + func (p AckPolicy) String() string + type ClientTrace struct + RequestSent func(subj string, payload []byte) + ResponseReceived func(subj string, payload []byte, hdr nats.Header) + type ClusterInfo struct + Leader string + Name string + Replicas []*PeerInfo + type ConsumeContext interface + Closed func() <-chan struct{} + Drain func() + Stop func() + type ConsumeErrHandlerFunc func(consumeCtx ConsumeContext, err error) + type Consumer interface + CachedInfo func() *ConsumerInfo + Consume func(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error) + Fetch func(batch int, opts ...FetchOpt) (MessageBatch, error) + FetchBytes func(maxBytes int, opts ...FetchOpt) (MessageBatch, error) + FetchNoWait func(batch int) (MessageBatch, error) + Info func(context.Context) (*ConsumerInfo, error) + Messages func(opts ...PullMessagesOpt) (MessagesContext, error) + Next func(opts ...FetchOpt) (Msg, error) + type ConsumerConfig struct + AckPolicy AckPolicy + AckWait time.Duration + BackOff []time.Duration + DeliverPolicy DeliverPolicy + Description string + Durable string + FilterSubject string + FilterSubjects []string + HeadersOnly bool + InactiveThreshold time.Duration + MaxAckPending int + MaxDeliver int + MaxRequestBatch int + MaxRequestExpires time.Duration + MaxRequestMaxBytes int + MaxWaiting int + MemoryStorage bool + Metadata map[string]string + Name string + OptStartSeq uint64 + OptStartTime *time.Time + RateLimit uint64 + ReplayPolicy ReplayPolicy + Replicas int + SampleFrequency string + type ConsumerInfo struct + AckFloor SequenceInfo + Cluster *ClusterInfo + Config ConsumerConfig + Created time.Time + Delivered SequenceInfo + Name string + NumAckPending int + NumPending uint64 + NumRedelivered int + NumWaiting int + PushBound bool + Stream string + TimeStamp time.Time + type ConsumerInfoLister interface + Err func() error + Info func() <-chan *ConsumerInfo + type ConsumerManager interface + Consumer func(ctx context.Context, consumer string) (Consumer, error) + ConsumerNames func(context.Context) ConsumerNameLister + CreateConsumer func(ctx context.Context, cfg ConsumerConfig) (Consumer, error) + CreateOrUpdateConsumer func(ctx context.Context, cfg ConsumerConfig) (Consumer, error) + DeleteConsumer func(ctx context.Context, consumer string) error + ListConsumers func(context.Context) ConsumerInfoLister + OrderedConsumer func(ctx context.Context, cfg OrderedConsumerConfig) (Consumer, error) + UpdateConsumer func(ctx context.Context, cfg ConsumerConfig) (Consumer, error) + type ConsumerNameLister interface + Err func() error + Name func() <-chan string + type DeleteMarkersOlderThan time.Duration + type DeliverPolicy int + const DeliverAllPolicy + const DeliverByStartSequencePolicy + const DeliverByStartTimePolicy + const DeliverLastPerSubjectPolicy + const DeliverLastPolicy + const DeliverNewPolicy + func (p *DeliverPolicy) UnmarshalJSON(data []byte) error + func (p DeliverPolicy) MarshalJSON() ([]byte, error) + func (p DeliverPolicy) String() string + type DiscardPolicy int + const DiscardNew + const DiscardOld + func (dp *DiscardPolicy) UnmarshalJSON(data []byte) error + func (dp DiscardPolicy) MarshalJSON() ([]byte, error) + func (dp DiscardPolicy) String() string + type ErrorCode uint16 + const JSErrCodeBadRequest + const JSErrCodeConsumerAlreadyExists + const JSErrCodeConsumerCreate + const JSErrCodeConsumerDoesNotExist + const JSErrCodeConsumerEmptyFilter + const JSErrCodeConsumerExists + const JSErrCodeConsumerNameExists + const JSErrCodeConsumerNotFound + const JSErrCodeDuplicateFilterSubjects + const JSErrCodeJetStreamNotEnabled + const JSErrCodeJetStreamNotEnabledForAccount + const JSErrCodeMessageNotFound + const JSErrCodeOverlappingFilterSubjects + const JSErrCodeStreamNameInUse + const JSErrCodeStreamNotFound + const JSErrCodeStreamWrongLastSequence + type ExternalStream struct + APIPrefix string + DeliverPrefix string + type FetchOpt func(*pullRequest) error + func FetchHeartbeat(hb time.Duration) FetchOpt + func FetchMaxWait(timeout time.Duration) FetchOpt + type GetMsgOpt func(*apiMsgGetRequest) error + func WithGetMsgSubject(subject string) GetMsgOpt + type GetObjectInfoOpt func(opts *getObjectInfoOpts) error + func GetObjectInfoShowDeleted() GetObjectInfoOpt + type GetObjectOpt func(opts *getObjectOpts) error + func GetObjectShowDeleted() GetObjectOpt + type JetStream interface + AccountInfo func(ctx context.Context) (*AccountInfo, error) + func New(nc *nats.Conn, opts ...JetStreamOpt) (JetStream, error) + func NewWithAPIPrefix(nc *nats.Conn, apiPrefix string, opts ...JetStreamOpt) (JetStream, error) + func NewWithDomain(nc *nats.Conn, domain string, opts ...JetStreamOpt) (JetStream, error) + type JetStreamError interface + APIError func() *APIError + var ErrAsyncPublishReplySubjectSet JetStreamError = &jsError{ ... } + var ErrBadBucket JetStreamError = &jsError{ ... } + var ErrBadObjectMeta JetStreamError = &jsError{ ... } + var ErrBadRequest JetStreamError = &jsError{ ... } + var ErrBucketExists JetStreamError = &jsError{ ... } + var ErrBucketMalformed JetStreamError = &jsError{ ... } + var ErrBucketNotFound JetStreamError = &jsError{ ... } + var ErrBucketRequired JetStreamError = &jsError{ ... } + var ErrCantGetBucket JetStreamError = &jsError{ ... } + var ErrConsumerCreate JetStreamError = &jsError{ ... } + var ErrConsumerDeleted JetStreamError = &jsError{ ... } + var ErrConsumerDoesNotExist JetStreamError = &jsError{ ... } + var ErrConsumerExists JetStreamError = &jsError{ ... } + var ErrConsumerHasActiveSubscription JetStreamError = &jsError{ ... } + var ErrConsumerLeadershipChanged JetStreamError = &jsError{ ... } + var ErrConsumerMultipleFilterSubjectsNotSupported JetStreamError = &jsError{ ... } + var ErrConsumerNameAlreadyInUse JetStreamError = &jsError{ ... } + var ErrConsumerNotFound JetStreamError = &jsError{ ... } + var ErrDigestMismatch JetStreamError = &jsError{ ... } + var ErrDuplicateFilterSubjects JetStreamError = &jsError{ ... } + var ErrEmptyFilter JetStreamError = &jsError{ ... } + var ErrEndOfData JetStreamError = &jsError{ ... } + var ErrHandlerRequired JetStreamError = &jsError{ ... } + var ErrHistoryTooLarge JetStreamError = &jsError{ ... } + var ErrInvalidBucketName JetStreamError = &jsError{ ... } + var ErrInvalidConsumerName JetStreamError = &jsError{ ... } + var ErrInvalidDigestFormat JetStreamError = &jsError{ ... } + var ErrInvalidJSAck JetStreamError = &jsError{ ... } + var ErrInvalidKey JetStreamError = &jsError{ ... } + var ErrInvalidOption JetStreamError = &jsError{ ... } + var ErrInvalidStoreName JetStreamError = &jsError{ ... } + var ErrInvalidStreamName JetStreamError = &jsError{ ... } + var ErrInvalidSubject JetStreamError = &jsError{ ... } + var ErrJetStreamNotEnabled JetStreamError = &jsError{ ... } + var ErrJetStreamNotEnabledForAccount JetStreamError = &jsError{ ... } + var ErrJetStreamPublisherClosed JetStreamError = &jsError{ ... } + var ErrKeyDeleted JetStreamError = &jsError{ ... } + var ErrKeyExists JetStreamError = &jsError{ ... } + var ErrKeyNotFound JetStreamError = &jsError{ ... } + var ErrKeyValueConfigRequired JetStreamError = &jsError{ ... } + var ErrLinkNotAllowed JetStreamError = &jsError{ ... } + var ErrMaxBytesExceeded JetStreamError = &jsError{ ... } + var ErrMsgAlreadyAckd JetStreamError = &jsError{ ... } + var ErrMsgDeleteUnsuccessful JetStreamError = &jsError{ ... } + var ErrMsgIteratorClosed JetStreamError = &jsError{ ... } + var ErrMsgNoReply JetStreamError = &jsError{ ... } + var ErrMsgNotBound JetStreamError = &jsError{ ... } + var ErrMsgNotFound JetStreamError = &jsError{ ... } + var ErrNameRequired JetStreamError = &jsError{ ... } + var ErrNoHeartbeat JetStreamError = &jsError{ ... } + var ErrNoKeysFound JetStreamError = &jsError{ ... } + var ErrNoLinkToDeleted JetStreamError = &jsError{ ... } + var ErrNoLinkToLink JetStreamError = &jsError{ ... } + var ErrNoMessages JetStreamError = &jsError{ ... } + var ErrNoObjectsFound JetStreamError = &jsError{ ... } + var ErrNoStreamResponse JetStreamError = &jsError{ ... } + var ErrNotJSMessage JetStreamError = &jsError{ ... } + var ErrObjectAlreadyExists JetStreamError = &jsError{ ... } + var ErrObjectConfigRequired JetStreamError = &jsError{ ... } + var ErrObjectNotFound JetStreamError = &jsError{ ... } + var ErrObjectRequired = &jsError{ ... } + var ErrOrderConsumerUsedAsConsume JetStreamError = &jsError{ ... } + var ErrOrderConsumerUsedAsFetch JetStreamError = &jsError{ ... } + var ErrOrderedConsumerConcurrentRequests JetStreamError = &jsError{ ... } + var ErrOrderedConsumerNotCreated JetStreamError = &jsError{ ... } + var ErrOrderedConsumerReset JetStreamError = &jsError{ ... } + var ErrOverlappingFilterSubjects JetStreamError = &jsError{ ... } + var ErrStreamNameAlreadyInUse JetStreamError = &jsError{ ... } + var ErrStreamNameRequired JetStreamError = &jsError{ ... } + var ErrStreamNotFound JetStreamError = &jsError{ ... } + var ErrStreamSourceMultipleFilterSubjectsNotSupported JetStreamError = &jsError{ ... } + var ErrStreamSourceNotSupported JetStreamError = &jsError{ ... } + var ErrStreamSourceSubjectTransformNotSupported JetStreamError = &jsError{ ... } + var ErrStreamSubjectTransformNotSupported JetStreamError = &jsError{ ... } + var ErrTooManyStalledMsgs JetStreamError = &jsError{ ... } + var ErrUpdateMetaDeleted JetStreamError = &jsError{ ... } + type JetStreamOpt func(*jsOpts) error + func WithClientTrace(ct *ClientTrace) JetStreamOpt + func WithPublishAsyncErrHandler(cb MsgErrHandler) JetStreamOpt + func WithPublishAsyncMaxPending(max int) JetStreamOpt + type KVDeleteOpt interface + func LastRevision(revision uint64) KVDeleteOpt + type KVPurgeOpt interface + type KeyLister interface + Keys func() <-chan string + Stop func() error + type KeyValue interface + Bucket func() string + Create func(ctx context.Context, key string, value []byte) (uint64, error) + Delete func(ctx context.Context, key string, opts ...KVDeleteOpt) error + Get func(ctx context.Context, key string) (KeyValueEntry, error) + GetRevision func(ctx context.Context, key string, revision uint64) (KeyValueEntry, error) + History func(ctx context.Context, key string, opts ...WatchOpt) ([]KeyValueEntry, error) + Keys func(ctx context.Context, opts ...WatchOpt) ([]string, error) + ListKeys func(ctx context.Context, opts ...WatchOpt) (KeyLister, error) + Purge func(ctx context.Context, key string, opts ...KVDeleteOpt) error + PurgeDeletes func(ctx context.Context, opts ...KVPurgeOpt) error + Put func(ctx context.Context, key string, value []byte) (uint64, error) + PutString func(ctx context.Context, key string, value string) (uint64, error) + Status func(ctx context.Context) (KeyValueStatus, error) + Update func(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) + Watch func(ctx context.Context, keys string, opts ...WatchOpt) (KeyWatcher, error) + WatchAll func(ctx context.Context, opts ...WatchOpt) (KeyWatcher, error) + type KeyValueBucketStatus struct + func (s *KeyValueBucketStatus) BackingStore() string + func (s *KeyValueBucketStatus) Bucket() string + func (s *KeyValueBucketStatus) Bytes() uint64 + func (s *KeyValueBucketStatus) History() int64 + func (s *KeyValueBucketStatus) IsCompressed() bool + func (s *KeyValueBucketStatus) StreamInfo() *StreamInfo + func (s *KeyValueBucketStatus) TTL() time.Duration + func (s *KeyValueBucketStatus) Values() uint64 + type KeyValueConfig struct + Bucket string + Compression bool + Description string + History uint8 + MaxBytes int64 + MaxValueSize int32 + Mirror *StreamSource + Placement *Placement + RePublish *RePublish + Replicas int + Sources []*StreamSource + Storage StorageType + TTL time.Duration + type KeyValueEntry interface + Bucket func() string + Created func() time.Time + Delta func() uint64 + Key func() string + Operation func() KeyValueOp + Revision func() uint64 + Value func() []byte + type KeyValueLister interface + Error func() error + Status func() <-chan KeyValueStatus + type KeyValueManager interface + CreateKeyValue func(ctx context.Context, cfg KeyValueConfig) (KeyValue, error) + CreateOrUpdateKeyValue func(ctx context.Context, cfg KeyValueConfig) (KeyValue, error) + DeleteKeyValue func(ctx context.Context, bucket string) error + KeyValue func(ctx context.Context, bucket string) (KeyValue, error) + KeyValueStoreNames func(ctx context.Context) KeyValueNamesLister + KeyValueStores func(ctx context.Context) KeyValueLister + UpdateKeyValue func(ctx context.Context, cfg KeyValueConfig) (KeyValue, error) + type KeyValueNamesLister interface + Error func() error + Name func() <-chan string + type KeyValueOp uint8 + const KeyValueDelete + const KeyValuePurge + const KeyValuePut + func (op KeyValueOp) String() string + type KeyValueStatus interface + BackingStore func() string + Bucket func() string + Bytes func() uint64 + History func() int64 + IsCompressed func() bool + TTL func() time.Duration + Values func() uint64 + type KeyWatcher interface + Stop func() error + Updates func() <-chan KeyValueEntry + type ListObjectsOpt func(opts *listObjectOpts) error + func ListObjectsShowDeleted() ListObjectsOpt + type MessageBatch interface + Error func() error + Messages func() <-chan Msg + type MessageHandler func(msg Msg) + type MessagesContext interface + Drain func() + Next func() (Msg, error) + Stop func() + type Msg interface + Ack func() error + Data func() []byte + DoubleAck func(context.Context) error + Headers func() nats.Header + InProgress func() error + Metadata func() (*MsgMetadata, error) + Nak func() error + NakWithDelay func(delay time.Duration) error + Reply func() string + Subject func() string + Term func() error + TermWithReason func(reason string) error + type MsgErrHandler func(JetStream, *nats.Msg, error) + type MsgMetadata struct + Consumer string + Domain string + NumDelivered uint64 + NumPending uint64 + Sequence SequencePair + Stream string + Timestamp time.Time + type ObjectBucketStatus struct + func (s *ObjectBucketStatus) BackingStore() string + func (s *ObjectBucketStatus) Bucket() string + func (s *ObjectBucketStatus) Description() string + func (s *ObjectBucketStatus) IsCompressed() bool + func (s *ObjectBucketStatus) Metadata() map[string]string + func (s *ObjectBucketStatus) Replicas() int + func (s *ObjectBucketStatus) Sealed() bool + func (s *ObjectBucketStatus) Size() uint64 + func (s *ObjectBucketStatus) Storage() StorageType + func (s *ObjectBucketStatus) StreamInfo() *StreamInfo + func (s *ObjectBucketStatus) TTL() time.Duration + type ObjectInfo struct + Bucket string + Chunks uint32 + Deleted bool + Digest string + ModTime time.Time + NUID string + Size uint64 + type ObjectLink struct + Bucket string + Name string + type ObjectMeta struct + Description string + Headers nats.Header + Metadata map[string]string + Name string + Opts *ObjectMetaOptions + type ObjectMetaOptions struct + ChunkSize uint32 + Link *ObjectLink + type ObjectResult interface + Error func() error + Info func() (*ObjectInfo, error) + type ObjectStore interface + AddBucketLink func(ctx context.Context, name string, bucket ObjectStore) (*ObjectInfo, error) + AddLink func(ctx context.Context, name string, obj *ObjectInfo) (*ObjectInfo, error) + Delete func(ctx context.Context, name string) error + Get func(ctx context.Context, name string, opts ...GetObjectOpt) (ObjectResult, error) + GetBytes func(ctx context.Context, name string, opts ...GetObjectOpt) ([]byte, error) + GetFile func(ctx context.Context, name, file string, opts ...GetObjectOpt) error + GetInfo func(ctx context.Context, name string, opts ...GetObjectInfoOpt) (*ObjectInfo, error) + GetString func(ctx context.Context, name string, opts ...GetObjectOpt) (string, error) + List func(ctx context.Context, opts ...ListObjectsOpt) ([]*ObjectInfo, error) + Put func(ctx context.Context, obj ObjectMeta, reader io.Reader) (*ObjectInfo, error) + PutBytes func(ctx context.Context, name string, data []byte) (*ObjectInfo, error) + PutFile func(ctx context.Context, file string) (*ObjectInfo, error) + PutString func(ctx context.Context, name string, data string) (*ObjectInfo, error) + Seal func(ctx context.Context) error + Status func(ctx context.Context) (ObjectStoreStatus, error) + UpdateMeta func(ctx context.Context, name string, meta ObjectMeta) error + Watch func(ctx context.Context, opts ...WatchOpt) (ObjectWatcher, error) + type ObjectStoreConfig struct + Bucket string + Compression bool + Description string + MaxBytes int64 + Metadata map[string]string + Placement *Placement + Replicas int + Storage StorageType + TTL time.Duration + type ObjectStoreManager interface + CreateObjectStore func(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) + CreateOrUpdateObjectStore func(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) + DeleteObjectStore func(ctx context.Context, bucket string) error + ObjectStore func(ctx context.Context, bucket string) (ObjectStore, error) + ObjectStoreNames func(ctx context.Context) ObjectStoreNamesLister + ObjectStores func(ctx context.Context) ObjectStoresLister + UpdateObjectStore func(ctx context.Context, cfg ObjectStoreConfig) (ObjectStore, error) + type ObjectStoreNamesLister interface + Error func() error + Name func() <-chan string + type ObjectStoreStatus interface + BackingStore func() string + Bucket func() string + Description func() string + IsCompressed func() bool + Metadata func() map[string]string + Replicas func() int + Sealed func() bool + Size func() uint64 + Storage func() StorageType + TTL func() time.Duration + type ObjectStoresLister interface + Error func() error + Status func() <-chan ObjectStoreStatus + type ObjectWatcher interface + Stop func() error + Updates func() <-chan *ObjectInfo + type OrderedConsumerConfig struct + DeliverPolicy DeliverPolicy + FilterSubjects []string + HeadersOnly bool + InactiveThreshold time.Duration + MaxResetAttempts int + OptStartSeq uint64 + OptStartTime *time.Time + ReplayPolicy ReplayPolicy + type PeerInfo struct + Active time.Duration + Current bool + Lag uint64 + Name string + Offline bool + type Placement struct + Cluster string + Tags []string + type PubAck struct + Domain string + Duplicate bool + Sequence uint64 + Stream string + type PubAckFuture interface + Err func() <-chan error + Msg func() *nats.Msg + Ok func() <-chan *PubAck + type PublishOpt func(*pubOpts) error + func WithExpectLastMsgID(id string) PublishOpt + func WithExpectLastSequence(seq uint64) PublishOpt + func WithExpectLastSequencePerSubject(seq uint64) PublishOpt + func WithExpectStream(stream string) PublishOpt + func WithMsgID(id string) PublishOpt + func WithRetryAttempts(num int) PublishOpt + func WithRetryWait(dur time.Duration) PublishOpt + func WithStallWait(ttl time.Duration) PublishOpt + type Publisher interface + CleanupPublisher func() + Publish func(ctx context.Context, subject string, payload []byte, opts ...PublishOpt) (*PubAck, error) + PublishAsync func(subject string, payload []byte, opts ...PublishOpt) (PubAckFuture, error) + PublishAsyncComplete func() <-chan struct{} + PublishAsyncPending func() int + PublishMsg func(ctx context.Context, msg *nats.Msg, opts ...PublishOpt) (*PubAck, error) + PublishMsgAsync func(msg *nats.Msg, opts ...PublishOpt) (PubAckFuture, error) + type PullConsumeOpt interface + func ConsumeErrHandler(cb ConsumeErrHandlerFunc) PullConsumeOpt + type PullExpiry time.Duration + type PullHeartbeat time.Duration + type PullMaxBytes int + type PullMaxMessages int + type PullMessagesOpt interface + func WithMessagesErrOnMissingHeartbeat(hbErr bool) PullMessagesOpt + type PullThresholdBytes int + type PullThresholdMessages int + type RawStreamMsg struct + Data []byte + Header nats.Header + Sequence uint64 + Subject string + Time time.Time + type RePublish struct + Destination string + HeadersOnly bool + Source string + type ReplayPolicy int + const ReplayInstantPolicy + const ReplayOriginalPolicy + func (p *ReplayPolicy) UnmarshalJSON(data []byte) error + func (p ReplayPolicy) MarshalJSON() ([]byte, error) + func (p ReplayPolicy) String() string + type RetentionPolicy int + const InterestPolicy + const LimitsPolicy + const WorkQueuePolicy + func (rp *RetentionPolicy) UnmarshalJSON(data []byte) error + func (rp RetentionPolicy) MarshalJSON() ([]byte, error) + func (rp RetentionPolicy) String() string + type SequenceInfo struct + Consumer uint64 + Last *time.Time + Stream uint64 + type SequencePair struct + Consumer uint64 + Stream uint64 + type StopAfter int + type StorageType int + const FileStorage + const MemoryStorage + func (st *StorageType) UnmarshalJSON(data []byte) error + func (st StorageType) MarshalJSON() ([]byte, error) + func (st StorageType) String() string + type StoreCompression uint8 + const NoCompression + const S2Compression + func (alg *StoreCompression) UnmarshalJSON(b []byte) error + func (alg StoreCompression) MarshalJSON() ([]byte, error) + func (alg StoreCompression) String() string + type Stream interface + CachedInfo func() *StreamInfo + DeleteMsg func(ctx context.Context, seq uint64) error + GetLastMsgForSubject func(ctx context.Context, subject string) (*RawStreamMsg, error) + GetMsg func(ctx context.Context, seq uint64, opts ...GetMsgOpt) (*RawStreamMsg, error) + Info func(ctx context.Context, opts ...StreamInfoOpt) (*StreamInfo, error) + Purge func(ctx context.Context, opts ...StreamPurgeOpt) error + SecureDeleteMsg func(ctx context.Context, seq uint64) error + type StreamConfig struct + AllowDirect bool + AllowRollup bool + Compression StoreCompression + ConsumerLimits StreamConsumerLimits + DenyDelete bool + DenyPurge bool + Description string + Discard DiscardPolicy + DiscardNewPerSubject bool + Duplicates time.Duration + FirstSeq uint64 + MaxAge time.Duration + MaxBytes int64 + MaxConsumers int + MaxMsgSize int32 + MaxMsgs int64 + MaxMsgsPerSubject int64 + Metadata map[string]string + Mirror *StreamSource + MirrorDirect bool + Name string + NoAck bool + Placement *Placement + RePublish *RePublish + Replicas int + Retention RetentionPolicy + Sealed bool + Sources []*StreamSource + Storage StorageType + SubjectTransform *SubjectTransformConfig + Subjects []string + Template string + type StreamConsumerLimits struct + InactiveThreshold time.Duration + MaxAckPending int + type StreamConsumerManager interface + Consumer func(ctx context.Context, stream string, consumer string) (Consumer, error) + CreateConsumer func(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error) + CreateOrUpdateConsumer func(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error) + DeleteConsumer func(ctx context.Context, stream string, consumer string) error + OrderedConsumer func(ctx context.Context, stream string, cfg OrderedConsumerConfig) (Consumer, error) + UpdateConsumer func(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error) + type StreamInfo struct + Cluster *ClusterInfo + Config StreamConfig + Created time.Time + Mirror *StreamSourceInfo + Sources []*StreamSourceInfo + State StreamState + TimeStamp time.Time + type StreamInfoLister interface + Err func() error + Info func() <-chan *StreamInfo + type StreamInfoOpt func(*streamInfoRequest) error + func WithDeletedDetails(deletedDetails bool) StreamInfoOpt + func WithSubjectFilter(subject string) StreamInfoOpt + type StreamListOpt func(*streamsRequest) error + func WithStreamListSubject(subject string) StreamListOpt + type StreamManager interface + CreateOrUpdateStream func(ctx context.Context, cfg StreamConfig) (Stream, error) + CreateStream func(ctx context.Context, cfg StreamConfig) (Stream, error) + DeleteStream func(ctx context.Context, stream string) error + ListStreams func(context.Context, ...StreamListOpt) StreamInfoLister + Stream func(ctx context.Context, stream string) (Stream, error) + StreamNameBySubject func(ctx context.Context, subject string) (string, error) + StreamNames func(context.Context, ...StreamListOpt) StreamNameLister + UpdateStream func(ctx context.Context, cfg StreamConfig) (Stream, error) + type StreamNameLister interface + Err func() error + Name func() <-chan string + type StreamPurgeOpt func(*StreamPurgeRequest) error + func WithPurgeKeep(keep uint64) StreamPurgeOpt + func WithPurgeSequence(sequence uint64) StreamPurgeOpt + func WithPurgeSubject(subject string) StreamPurgeOpt + type StreamPurgeRequest struct + Keep uint64 + Sequence uint64 + Subject string + type StreamSource struct + Domain string + External *ExternalStream + FilterSubject string + Name string + OptStartSeq uint64 + OptStartTime *time.Time + SubjectTransforms []SubjectTransformConfig + type StreamSourceInfo struct + Active time.Duration + FilterSubject string + Lag uint64 + Name string + SubjectTransforms []SubjectTransformConfig + type StreamState struct + Bytes uint64 + Consumers int + Deleted []uint64 + FirstSeq uint64 + FirstTime time.Time + LastSeq uint64 + LastTime time.Time + Msgs uint64 + NumDeleted int + NumSubjects uint64 + Subjects map[string]uint64 + type SubjectTransformConfig struct + Destination string + Source string + type Tier struct + Consumers int + Limits AccountLimits + Memory uint64 + ReservedMemory uint64 + ReservedStore uint64 + Store uint64 + Streams int + type WatchOpt interface + func IgnoreDeletes() WatchOpt + func IncludeHistory() WatchOpt + func MetaOnly() WatchOpt + func ResumeFromRevision(revision uint64) WatchOpt + func UpdatesOnly() WatchOpt
Go to main page