nats

Apr 14, 2026

Changes in this version

+ const DefaultWriteBufSize

type Conn

+ func (nc *Conn) ConnectedServerJetStream() (bool, int)

+ func (nc *Conn) IsSystemAccount() bool

type Option

+ func ReconnectOnFlusherError() Option

+ func WriteBufferSize(size int) Option

type Options

+ ReconnectOnFlusherError bool

+ WriteBufferSize int

type ServerInfo

+ IsSystemAccount bool

+ JSApiLevel int

+ JetStream bool

Mar 25, 2026

Changes in this version

+ var ErrTLS = errors.New("nats: tls error")

type APIStats

+ Inflight uint64

+ Level int

Feb 23, 2026

Changes in this version

+ var ErrMixingWebsocketSchemes = errors.New("nats: mixing of websocket and non websocket URLs is not allowed")

+ var ErrServerNotInPool = errors.New("nats: selected server is not in the pool")

type Conn

+ func (nc *Conn) ServerPool() []Server

+ func (nc *Conn) SetServerPool(servers []string) error

type KeyValueBucketStatus

+ func (s *KeyValueBucketStatus) Config() KeyValueConfig

type KeyValueStatus

+ Config func() KeyValueConfig

type Option

+ func IgnoreDiscoveredServers() Option

+ func ReconnectToServer(cb ReconnectToServerHandler) Option

type Options

+ IgnoreDiscoveredServers bool

+ ReconnectToServerCB ReconnectToServerHandler

+ type ReconnectToServerHandler func([]Server, ServerInfo) (*Server, time.Duration)

+ type Server struct

+ Reconnects int

+ URL *url.URL

+ type ServerInfo struct

+ AuthRequired bool

+ CID uint64

+ ClientIP string

+ Cluster string

+ ConnectURLs []string

+ Headers bool

+ Host string

+ ID string

+ LameDuckMode bool

+ MaxPayload int64

+ Name string

+ Nonce string

+ Port int

+ Proto int

+ TLSAvailable bool

+ TLSRequired bool

+ Version string

Dec 17, 2025

Changes in this version

type Option

+ func SkipSubjectValidation() Option

type Options

+ SkipSubjectValidation bool

Oct 14, 2025

Changes in this version

+ var ErrWebSocketHeadersAlreadySet = errors.New("nats: websocket connection headers already set")

type Option

+ func WebSocketConnectionHeaders(headers http.Header) Option

+ func WebSocketConnectionHeadersHandler(cb WebSocketHeadersHandler) Option

type Options

+ WebSocketConnectionHeaders http.Header

+ WebSocketConnectionHeadersHandler WebSocketHeadersHandler

+ type WebSocketHeadersHandler func() (http.Header, error)

Sep 22, 2025

Changes in this version

type ClusterInfo

+ LeaderSince *time.Time

+ RaftGroup string

+ SystemAcc bool

+ TrafficAcc string

type JetStreamError

+ var ErrConsumerInfoOnOrderedReset JetStreamError = &jsError{ ... }

Aug 20, 2025

Changes in this version

+ const MAX_ACCOUNT_CONNECTIONS_ERR

+ var ErrMaxAccountConnectionsExceeded = errors.New("nats: maximum account active connections exceeded")

type JetStreamError

+ var ErrConsumerCreationResponseEmpty JetStreamError = &jsError{ ... }

Jul 29, 2025

Changes in this version

+ var ErrKeyWatcherTimeout = errors.New("nats: key watcher timed out waiting for initial keys")

type KeyLister

+ Error func() <-chan error

type KeyWatcher

+ Error func() <-chan error

type Option

+ func UserCredentialBytes(userOrChainedFileBytes []byte, seedFiles ...[]byte) Option

Jun 3, 2025

Changes in this version

type Conn

+ func (nc *Conn) LocalAddr() string

Apr 17, 2025

Changes in this version

type Conn

+ func (nc *Conn) RemoveStatusListener(ch chan (Status))

Apr 3, 2025

Changes in this version

type JetStreamError

+ var ErrFetchDisconnected = &jsError{ ... }

Mar 24, 2025

Changes in this version

+ const MsgTTLHdr

type ErrorCode

+ const JSErrCodeJetStreamNotAvailable

type JSOpt

+ func PublishAsyncTimeout(dur time.Duration) JSOpt

type JetStreamError

+ var ErrAsyncPublishTimeout JetStreamError = &jsError{ ... }

+ var ErrTooManyStalledMsgs JetStreamError = &jsError{ ... }

type Option

+ func ReconnectErrHandler(cb ConnErrHandler) Option

type Options

+ ReconnectErrCB ConnErrHandler

type PubOpt

+ func MsgTTL(dur time.Duration) PubOpt

type StreamConfig

+ AllowMsgTTL bool

+ SubjectDeleteMarkerTTL time.Duration

Dec 17, 2024

Changes in this version

+ const MAX_SUBSCRIPTIONS_ERR

+ var ErrMaxSubscriptionsExceeded = errors.New("nats: server maximum subscriptions exceeded")

+ var ErrPermissionViolation = errors.New("nats: permissions violation")

+ var ErrUserInfoAlreadySet = errors.New("nats: cannot set user info callback and user/pass")

type KeyValue

+ WatchFiltered func(keys []string, opts ...WatchOpt) (KeyWatcher, error)

type Option

+ func PermissionErrOnSubscribe(enabled bool) Option

+ func UserInfoHandler(cb UserInfoCB) Option

type Options

+ PermissionErrOnSubscribe bool

+ UserInfo UserInfoCB

type Subscription

+ func (sub *Subscription) Msgs() iter.Seq2[*Msg, error]

+ func (sub *Subscription) MsgsTimeout(timeout time.Duration) iter.Seq2[*Msg, error]

+ type UserInfoCB func() (string, string)

Aug 13, 2024

Changes in this version

type JetStream

+ CleanupPublisher func()

type JetStreamError

+ var ErrJetStreamPublisherClosed JetStreamError = &jsError{ ... }

May 17, 2024

Changes in this version

type Conn

+ func (nc *Conn) ForceReconnect() error

type JetStreamError

+ var ErrInvalidFilterSubject JetStreamError = &jsError{ ... }

Mar 20, 2024

Changes in this version

+ const SubscriptionActive

+ const SubscriptionClosed

+ const SubscriptionDraining

+ const SubscriptionSlowConsumer

+ var ErrClientCertOrRootCAsRequired = errors.New("nats: at least one of certCB or rootCAsCB must be set")

type JetStreamError

+ var ErrSubscriptionClosed JetStreamError = &jsError{ ... }

type Option

+ func ClientTLSConfig(certCB TLSCertHandler, rootCAsCB RootCAsHandler) Option

+ type SubStatus int

+ func (s SubStatus) String() string

type Subscription

+ func (s *Subscription) IsDraining() bool

+ func (s *Subscription) StatusChanged(statuses ...SubStatus) <-chan SubStatus

Feb 14, 2024

Changes in this version

type ObjectBucketStatus

+ func (s *ObjectBucketStatus) IsCompressed() bool

type ObjectStoreConfig

+ Compression bool

type ObjectStoreStatus

+ IsCompressed func() bool

Jan 12, 2024

Changes in this version

+ type KeyLister interface

+ Keys func() <-chan string

+ Stop func() error

type KeyValue

+ ListKeys func(opts ...WatchOpt) (KeyLister, error)

type KeyValueBucketStatus

+ func (s *KeyValueBucketStatus) IsCompressed() bool

type KeyValueConfig

+ Compression bool

type KeyValueStatus

+ IsCompressed func() bool

type Tier

+ ReservedMemory uint64

+ ReservedStore uint64

Oct 15, 2023

Changes in this version

type Option

+ func TLSHandshakeFirst() Option

type Options

+ TLSHandshakeFirst bool

Sep 20, 2023

Changes in this version

type ConsumerConfig

+ FilterSubjects []string

+ Metadata map[string]string

type ErrorCode

+ const JSErrCodeConsumerEmptyFilter

+ const JSErrCodeDuplicateFilterSubjects

+ const JSErrCodeOverlappingFilterSubjects

+ const JSStreamInvalidConfig

type JetStreamError

+ var ErrConsumerMultipleFilterSubjectsNotSupported JetStreamError = &jsError{ ... }

+ var ErrDuplicateFilterSubjects JetStreamError = &jsError{ ... }

+ var ErrEmptyFilter JetStreamError = &jsError{ ... }

+ var ErrNoHeartbeat JetStreamError = &jsError{ ... }

+ var ErrOverlappingFilterSubjects JetStreamError = &jsError{ ... }

+ var ErrStreamSourceMultipleSubjectTransformsNotSupported JetStreamError = &jsError{ ... }

+ var ErrStreamSourceNotSupported JetStreamError = &jsError{ ... }

+ var ErrStreamSourceSubjectTransformNotSupported JetStreamError = &jsError{ ... }

+ var ErrStreamSubjectTransformNotSupported JetStreamError = &jsError{ ... }

type ObjectBucketStatus

+ func (s *ObjectBucketStatus) Metadata() map[string]string

type ObjectMeta

+ Metadata map[string]string

type ObjectStoreConfig

+ Metadata map[string]string

type ObjectStoreStatus

+ Metadata func() map[string]string

+ type PullHeartbeat time.Duration

+ 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 StreamConfig

+ Compression StoreCompression

+ ConsumerLimits StreamConsumerLimits

+ FirstSeq uint64

+ Metadata map[string]string

+ SubjectTransform *SubjectTransformConfig

+ type StreamConsumerLimits struct

+ InactiveThreshold time.Duration

+ MaxAckPending int

type StreamSource

+ SubjectTransforms []SubjectTransformConfig

type StreamSourceInfo

+ FilterSubject string

+ SubjectTransforms []SubjectTransformConfig

type SubOpt

+ func ConsumerFilterSubjects(subjects ...string) SubOpt

+ type SubjectTransformConfig struct

+ Destination string

+ Source string

Sep 13, 2023

Changes in this version

+ const DefaultFlusherTimeout

type Subscription

+ func (s *Subscription) SetClosedHandler(handler func(subject string))

+ func (sub *Subscription) InitialConsumerPending() (uint64, error)

type WatchOpt

+ func UpdatesOnly() WatchOpt

May 23, 2023

Changes in this version

type Conn

+ func (nc *Conn) StatusChanged(statuses ...Status) chan Status

type Header

+ func DecodeHeadersMsg(data []byte) (Header, error)

type Msg

+ func (m *Msg) Size() int

type Options

+ RootCAsCB RootCAsHandler

+ TLSCertCB TLSCertHandler

+ type RootCAsHandler func() (*x509.CertPool, error)

type SubOpt

+ func ConsumerName(name string) SubOpt

+ func SkipConsumerLookup() SubOpt

+ type TLSCertHandler func() (tls.Certificate, error)

Feb 24, 2023

Changes in this version

type ErrorCode

+ const JSErrCodeInsufficientResourcesErr

+ type MessageBatch interface

+ Done func() <-chan struct{}

+ Error func() error

+ Messages func() <-chan *Msg

type Option

+ func SkipHostLookup() Option

type Options

+ SkipHostLookup bool

type Subscription

+ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, error)

Dec 20, 2022

Changes in this version

type Conn

+ func (nc *Conn) ClosedHandler() ConnHandler

+ func (nc *Conn) DisconnectErrHandler() ConnErrHandler

+ func (nc *Conn) DiscoveredServersHandler() ConnHandler

+ func (nc *Conn) ErrorHandler() ErrHandler

+ func (nc *Conn) ReconnectHandler() ConnHandler

Dec 5, 2022

Changes in this version

type ErrorCode

+ const JSErrCodeStreamWrongLastSequence

type JetStreamError

+ var ErrConsumerLeadershipChanged JetStreamError = &jsError{ ... }

+ var ErrKeyExists JetStreamError = &jsError{ ... }

type Option

+ func ConnectHandler(cb ConnHandler) Option

+ func IgnoreAuthErrorAbort() Option

type Options

+ ConnectedCB ConnHandler

+ IgnoreAuthErrorAbort bool

Nov 11, 2022

Changes in this version

type JetStreamError

+ var ErrConsumerDeleted JetStreamError = &jsError{ ... }

Nov 5, 2022

Changes in this version

type JetStreamManager

+ StreamNameBySubject func(string, ...JSOpt) (string, error)

+ type StreamAlternate struct

+ Cluster string

+ Domain string

+ Name string

type StreamInfo

+ Alternates []*StreamAlternate

Oct 27, 2022

Changes in this version

type KeyValueConfig

+ Mirror *StreamSource

+ Sources []*StreamSource

type StreamSource

+ Domain string

Oct 12, 2022

Changes in this version

+ type GetObjectInfoOpt interface

+ func GetObjectInfoShowDeleted() GetObjectInfoOpt

+ type GetObjectOpt interface

+ func GetObjectShowDeleted() GetObjectOpt

type KeyValueBucketStatus

+ func (s *KeyValueBucketStatus) Bytes() uint64

type KeyValueManager

+ KeyValueStoreNames func() <-chan string

+ KeyValueStores func() <-chan KeyValueStatus

type KeyValueStatus

+ Bytes func() uint64

+ type ListObjectsOpt interface

+ func ListObjectsShowDeleted() ListObjectsOpt

type StreamConfig

+ DiscardNewPerSubject bool

Sep 16, 2022

Changes in this version

+ const JSSubject

+ const JSTimeStamp

+ var ErrBucketMalformed = errors.New("nats: bucket malformed")

+ var ErrBucketRequired = errors.New("nats: bucket required")

+ var ErrCantGetBucket = errors.New("nats: invalid Get, object is a link to a bucket")

+ var ErrConnectionNotTLS = errors.New("nats: connection is not tls")

+ var ErrInvalidDigestFormat = errors.New("nats: object digest hash has invalid format")

+ var ErrLinkNotAllowed = errors.New("nats: link cannot be set when putting the object in bucket")

+ var ErrNameRequired = errors.New("nats: name is required")

+ var ErrNeeds262 = errors.New("nats: object-store requires at least server version 2.6.2")

+ var ErrNoLinkToDeleted = errors.New("nats: not allowed to link to a deleted object")

+ var ErrNoLinkToLink = errors.New("nats: not allowed to link to another link")

+ var ErrObjectAlreadyExists = errors.New("nats: an object already exists with that name")

+ var ErrObjectRequired = errors.New("nats: object required")

+ var ErrUpdateMetaDeleted = errors.New("nats: cannot update meta for a deleted object")

+ 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 AccountInfo

+ Tiers map[string]Tier

type AccountLimits

+ MaxAckPending int

+ MaxBytesRequired bool

+ MemoryMaxStreamBytes int64

+ StoreMaxStreamBytes int64

type Conn

+ func (nc *Conn) NewInbox() string

+ func (nc *Conn) TLSConnectionState() (tls.ConnectionState, error)

type ConsumerConfig

+ MaxRequestMaxBytes int

+ Name string

+ type ErrorCode uint16

+ const JSErrCodeBadRequest

+ const JSErrCodeConsumerAlreadyExists

+ const JSErrCodeConsumerNameExists

+ const JSErrCodeConsumerNotFound

+ const JSErrCodeJetStreamNotEnabled

+ const JSErrCodeJetStreamNotEnabledForAccount

+ const JSErrCodeMessageNotFound

+ const JSErrCodeStreamNameInUse

+ const JSErrCodeStreamNotFound

+ type InProcessConnProvider interface

+ InProcessConn func() (net.Conn, error)

type JSOpt

+ func DirectGet() JSOpt

+ func DirectGetNext(subject string) JSOpt

+ func StreamListFilter(subject string) JSOpt

+ func UseLegacyDurableConsumers() JSOpt

+ type JetStreamError interface

+ APIError func() *APIError

+ var ErrBadRequest JetStreamError = &jsError{ ... }

+ var ErrCantAckIfConsumerAckNone JetStreamError = &jsError{ ... }

+ var ErrConsumerNameAlreadyInUse JetStreamError = &jsError{ ... }

+ var ErrJetStreamNotEnabledForAccount JetStreamError = &jsError{ ... }

type JetStreamManager

+ Consumers func(stream string, opts ...JSOpt) <-chan *ConsumerInfo

+ GetLastMsg func(name, subject string, opts ...JSOpt) (*RawStreamMsg, error)

+ SecureDeleteMsg func(name string, seq uint64, opts ...JSOpt) error

+ Streams func(opts ...JSOpt) <-chan *StreamInfo

type KeyValueConfig

+ RePublish *RePublish

type Msg

+ func (m *Msg) Equal(msg *Msg) bool

type ObjectStoreManager

+ ObjectStoreNames func(opts ...ObjectOpt) <-chan string

+ ObjectStores func(opts ...ObjectOpt) <-chan ObjectStore

type Option

+ func InProcessServer(server InProcessConnProvider) Option

+ func UserJWTAndSeed(jwt string, seed string) Option

type Options

+ InProcessServer InProcessConnProvider

+ type PullMaxBytes int

+ type RePublish struct

+ Destination string

+ HeadersOnly bool

+ Source string

type StreamConfig

+ AllowDirect bool

+ MirrorDirect bool

+ type StreamInfoRequest struct

+ DeletedDetails bool

+ SubjectsFilter string

+ type StreamPurgeRequest struct

+ Keep uint64

+ Sequence uint64

+ Subject string

type StreamSourceInfo

+ Error *APIError

+ External *ExternalStream

type StreamState

+ Deleted []uint64

+ NumDeleted int

+ NumSubjects uint64

+ Subjects map[string]uint64

type SubOpt

+ func ConsumerMemoryStorage() SubOpt

+ func ConsumerReplicas(replicas int) SubOpt

+ func MaxRequestMaxBytes(bytes int) SubOpt

+ type Tier struct

+ Consumers int

+ Limits AccountLimits

+ Memory uint64

+ Store uint64

+ Streams int

May 26, 2022

Changes in this version

+ const JSLastSequence

+ const JSSequence

+ const JSStream

type ConsumerConfig

+ MemoryStorage bool

+ Replicas int

type Option

+ func ProxyPath(path string) Option

type Options

+ ProxyPath string

type StreamConfig

+ RePublish *SubjectMapping

+ type SubjectMapping struct

+ Destination string

+ Source string

May 4, 2022

Changes in this version

type ObjectStoreConfig

+ MaxBytes int64

Apr 8, 2022

Changes in this version

+ const DefaultPubRetryAttempts

+ const DefaultPubRetryWait

+ const MAX_CONNECTIONS_ERR

+ var ErrInvalidConsumerName = errors.New("nats: invalid consumer name")

+ var ErrMaxConnectionsExceeded = errors.New("nats: server maximum connections exceeded")

+ var ErrMsgAlreadyAckd = errors.New("nats: message was already acknowledged")

+ var ErrStreamConfigRequired = errors.New("nats: stream configuration is required")

+ var ErrStreamInfoMaxSubjects = errors.New("nats: subject details would exceed maximum allowed")

+ var ErrStreamNameAlreadyInUse = errors.New("nats: stream name already in use")

+ type ClientTrace struct

+ RequestSent func(subj string, payload []byte)

+ ResponseReceived func(subj string, payload []byte, hdr Header)

type Conn

+ func (nc *Conn) ConnectedUrlRedacted() string

type ConsumerConfig

+ BackOff []time.Duration

+ InactiveThreshold time.Duration

+ MaxRequestBatch int

+ MaxRequestExpires time.Duration

+ type DeleteMarkersOlderThan time.Duration

+ type DeleteOpt interface

+ func LastRevision(revision uint64) DeleteOpt

type JetStreamManager

+ UpdateConsumer func(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error)

type KeyValue

+ GetRevision func(key string, revision uint64) (entry KeyValueEntry, err error)

+ Status func() (KeyValueStatus, error)

+ type KeyValueBucketStatus struct

+ func (s *KeyValueBucketStatus) BackingStore() string

+ func (s *KeyValueBucketStatus) Bucket() string

+ func (s *KeyValueBucketStatus) History() int64

+ func (s *KeyValueBucketStatus) StreamInfo() *StreamInfo

+ func (s *KeyValueBucketStatus) TTL() time.Duration

+ func (s *KeyValueBucketStatus) Values() uint64

type KeyValueConfig

+ Placement *Placement

+ type KeyValueStatus interface

+ BackingStore func() string

+ Bucket func() string

+ History func() int64

+ TTL func() time.Duration

+ Values func() uint64

type KeyWatcher

+ Context func() context.Context

type Msg

+ func (m *Msg) NakWithDelay(delay time.Duration, opts ...AckOpt) error

+ type ObjectBucketStatus struct

+ func (s *ObjectBucketStatus) BackingStore() string

+ func (s *ObjectBucketStatus) Bucket() string

+ func (s *ObjectBucketStatus) Description() 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 ObjectStore

+ Status func() (ObjectStoreStatus, error)

type ObjectStoreConfig

+ Placement *Placement

+ type ObjectStoreStatus interface

+ BackingStore func() string

+ Bucket func() string

+ Description func() string

+ Replicas func() int

+ Sealed func() bool

+ Size func() uint64

+ Storage func() StorageType

+ TTL func() time.Duration

type PubOpt

+ func RetryAttempts(num int) PubOpt

+ func RetryWait(dur time.Duration) PubOpt

+ func StallWait(ttl time.Duration) PubOpt

+ type PurgeOpt interface

type SubOpt

+ func BackOff(backOff []time.Duration) SubOpt

+ func InactiveThreshold(threshold time.Duration) SubOpt

+ func MaxRequestBatch(max int) SubOpt

+ func MaxRequestExpires(max time.Duration) SubOpt

type WatchOpt

+ func MetaOnly() WatchOpt

Oct 7, 2021

Changes in this version

+ const AllKeys

+ const KeyValueMaxHistory

+ const MsgRollup

+ const MsgRollupAll

+ const MsgRollupSubject

+ const MsgSize

+ var ErrBadBucket = errors.New("nats: bucket not valid key-value store")

+ var ErrBadObjectMeta = errors.New("nats: object-store meta information invalid")

+ var ErrBucketNotFound = errors.New("nats: bucket not found")

+ var ErrDigestMismatch = errors.New("nats: received a corrupt object, digests do not match")

+ var ErrHistoryToLarge = errors.New("nats: history limited to a max of 64")

+ var ErrInvalidBucketName = errors.New("nats: invalid bucket name")

+ var ErrInvalidKey = errors.New("nats: invalid key")

+ var ErrInvalidObjectName = errors.New("nats: invalid object name")

+ var ErrInvalidStoreName = errors.New("nats: invalid object-store name")

+ var ErrKeyDeleted = errors.New("nats: key was deleted")

+ var ErrKeyNotFound = errors.New("nats: key not found")

+ var ErrKeyValueConfigRequired = errors.New("nats: config required")

+ var ErrMsgNotFound = errors.New("nats: message not found")

+ var ErrNoKeysFound = errors.New("nats: no keys found")

+ var ErrNoObjectsFound = errors.New("nats: no objects found")

+ var ErrObjectConfigRequired = errors.New("nats: object-store config required")

+ var ErrObjectNotFound = errors.New("nats: object not found")

type Conn

+ func (nc *Conn) ConnectedServerVersion() string

type ConsumerConfig

+ HeadersOnly bool

+ type KeyValue interface

+ Bucket func() string

+ Create func(key string, value []byte) (revision uint64, err error)

+ Delete func(key string) error

+ Get func(key string) (entry KeyValueEntry, err error)

+ History func(key string, opts ...WatchOpt) ([]KeyValueEntry, error)

+ Keys func(opts ...WatchOpt) ([]string, error)

+ Purge func(key string) error

+ PurgeDeletes func(opts ...WatchOpt) error

+ Put func(key string, value []byte) (revision uint64, err error)

+ PutString func(key string, value string) (revision uint64, err error)

+ Update func(key string, value []byte, last uint64) (revision uint64, err error)

+ Watch func(keys string, opts ...WatchOpt) (KeyWatcher, error)

+ WatchAll func(opts ...WatchOpt) (KeyWatcher, error)

+ type KeyValueConfig struct

+ Bucket string

+ Description string

+ History uint8

+ MaxBytes int64

+ MaxValueSize int32

+ Replicas int

+ 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 KeyValueManager interface

+ CreateKeyValue func(cfg *KeyValueConfig) (KeyValue, error)

+ DeleteKeyValue func(bucket string) error

+ KeyValue func(bucket string) (KeyValue, error)

+ type KeyValueOp uint8

+ const KeyValueDelete

+ const KeyValuePurge

+ const KeyValuePut

+ func (op KeyValueOp) String() string

+ type KeyWatcher interface

+ Stop func() error

+ Updates func() <-chan KeyValueEntry

+ 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 Header

+ Name string

+ Opts *ObjectMetaOptions

+ type ObjectMetaOptions struct

+ ChunkSize uint32

+ Link *ObjectLink

+ type ObjectOpt interface

+ type ObjectResult interface

+ Error func() error

+ Info func() (*ObjectInfo, error)

+ type ObjectStore interface

+ AddBucketLink func(name string, bucket ObjectStore) (*ObjectInfo, error)

+ AddLink func(name string, obj *ObjectInfo) (*ObjectInfo, error)

+ Delete func(name string) error

+ Get func(name string, opts ...ObjectOpt) (ObjectResult, error)

+ GetBytes func(name string, opts ...ObjectOpt) ([]byte, error)

+ GetFile func(name, file string, opts ...ObjectOpt) error

+ GetInfo func(name string) (*ObjectInfo, error)

+ GetString func(name string, opts ...ObjectOpt) (string, error)

+ List func(opts ...WatchOpt) ([]*ObjectInfo, error)

+ Put func(obj *ObjectMeta, reader io.Reader, opts ...ObjectOpt) (*ObjectInfo, error)

+ PutBytes func(name string, data []byte, opts ...ObjectOpt) (*ObjectInfo, error)

+ PutFile func(file string, opts ...ObjectOpt) (*ObjectInfo, error)

+ PutString func(name string, data string, opts ...ObjectOpt) (*ObjectInfo, error)

+ Seal func() error

+ UpdateMeta func(name string, meta *ObjectMeta) error

+ Watch func(opts ...WatchOpt) (ObjectWatcher, error)

+ type ObjectStoreConfig struct

+ Bucket string

+ Description string

+ Replicas int

+ Storage StorageType

+ TTL time.Duration

+ type ObjectStoreManager interface

+ CreateObjectStore func(cfg *ObjectStoreConfig) (ObjectStore, error)

+ DeleteObjectStore func(bucket string) error

+ ObjectStore func(bucket string) (ObjectStore, error)

+ type ObjectWatcher interface

+ Stop func() error

+ Updates func() <-chan *ObjectInfo

type StreamConfig

+ AllowRollup bool

+ DenyDelete bool

+ DenyPurge bool

+ Sealed bool

type SubOpt

+ func HeadersOnly() SubOpt

+ type WatchOpt interface

+ func IgnoreDeletes() WatchOpt

+ func IncludeHistory() WatchOpt

Sep 3, 2021

Changes in this version

type Status

+ func (s Status) String() string

Aug 26, 2021

Changes in this version

+ const ExpectedLastSubjSeqHdr

+ var ErrConsumerNameRequired = errors.New("nats: consumer name is required")

+ var ErrConsumerNotActive = errors.New("nats: consumer not active")

+ var ErrConsumerNotFound = errors.New("nats: consumer not found")

+ var ErrPullSubscribeRequired = errors.New("nats: must use pull subscribe to bind to pull based consumer")

+ var ErrPullSubscribeToPushConsumer = errors.New("nats: cannot pull subscribe to push based consumer")

+ var ErrStreamNotFound = errors.New("nats: stream not found")

type AccountInfo

+ Domain string

type ConsumerConfig

+ DeliverGroup string

+ Description string

type ConsumerInfo

+ PushBound bool

type DeliverPolicy

+ const DeliverLastPerSubjectPolicy

type JSOpt

+ func Domain(domain string) JSOpt

type JetStream

+ ChanQueueSubscribe func(subj, queue string, ch chan *Msg, opts ...SubOpt) (*Subscription, error)

type MsgMetadata

+ Domain string

type Option

+ func CustomInboxPrefix(p string) Option

type Options

+ InboxPrefix string

type PubAck

+ Domain string

type PubOpt

+ func ExpectLastSequencePerSubject(seq uint64) PubOpt

+ type SequenceInfo struct

+ Consumer uint64

+ Last *time.Time

+ Stream uint64

type StreamConfig

+ Description string

+ MaxMsgsPerSubject int64

type SubOpt

+ func Bind(stream, consumer string) SubOpt

+ func DeliverLastPerSubject() SubOpt

+ func DeliverSubject(subject string) SubOpt

+ func Description(description string) SubOpt

+ func OrderedConsumer() SubOpt

+ func ReplayInstant() SubOpt

May 4, 2021

Changes in this version

+ const ACCOUNT_AUTHENTICATION_EXPIRED_ERR

+ const AUTHENTICATION_REVOKED_ERR

+ const ExpectedLastMsgIdHdr

+ const ExpectedLastSeqHdr

+ const ExpectedStreamHdr

+ const MsgIdHdr

+ const OP_H

+ const PullSubscription

+ var ErrAccountAuthExpired = errors.New("nats: account authentication expired")

+ var ErrAuthRevoked = errors.New("nats: authentication revoked")

+ var ErrBadHeaderMsg = errors.New("nats: message could not decode headers")

+ var ErrConsumerConfigRequired = errors.New("nats: consumer configuration is required")

+ var ErrContextAndTimeout = errors.New("nats: context and timeout can not both be set")

+ var ErrDeliverSubjectRequired = errors.New("nats: deliver subject is required")

+ var ErrHeadersNotSupported = errors.New("nats: headers not supported by this server")

+ var ErrInvalidDurableName = errors.New("nats: invalid durable name")

+ var ErrInvalidJSAck = errors.New("nats: invalid jetstream publish response")

+ var ErrInvalidStreamName = errors.New("nats: invalid stream name")

+ var ErrJetStreamBadPre = errors.New("nats: jetstream api prefix not valid")

+ var ErrJetStreamNotEnabled = errors.New("nats: jetstream not enabled")

+ var ErrMultiStreamUnsupported = errors.New("nats: multiple streams are not supported")

+ var ErrNoContextOrTimeout = errors.New("nats: no context or timeout given")

+ var ErrNoMatchingStream = errors.New("nats: no stream matches subject")

+ var ErrNoResponders = errors.New("nats: no responders available for request")

+ var ErrNoStreamResponse = errors.New("nats: no response from stream")

+ var ErrNotJSMessage = errors.New("nats: not a jetstream message")

+ var ErrPullModeNotAllowed = errors.New("nats: pull based not supported")

+ var ErrStreamNameRequired = errors.New("nats: stream name is required")

+ var ErrStreamSnapshotConfigRequired = errors.New("nats: stream snapshot configuration is required")

+ var ErrSubjectMismatch = errors.New("nats: subject does not match consumer")

+ type APIStats struct

+ Errors uint64

+ Total uint64

+ type AccountInfo struct

+ API APIStats

+ Consumers int

+ Limits AccountLimits

+ Memory uint64

+ Store uint64

+ Streams int

+ type AccountLimits struct

+ MaxConsumers int

+ MaxMemory int64

+ MaxStore int64

+ MaxStreams int

+ type AckOpt interface

+ 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 AckWait time.Duration

+ type ClusterInfo struct

+ Leader string

+ Name string

+ Replicas []*PeerInfo

type Conn

+ func (nc *Conn) ConnectedClusterName() string

+ func (nc *Conn) ConnectedServerName() string

+ func (nc *Conn) HeadersSupported() bool

+ func (nc *Conn) JetStream(opts ...JSOpt) (JetStreamContext, error)

+ func (nc *Conn) RequestMsg(msg *Msg, timeout time.Duration) (*Msg, error)

+ func (nc *Conn) RequestMsgWithContext(ctx context.Context, msg *Msg) (*Msg, error)

+ type ConsumerConfig struct

+ AckPolicy AckPolicy

+ AckWait time.Duration

+ DeliverPolicy DeliverPolicy

+ DeliverSubject string

+ Durable string

+ FilterSubject string

+ FlowControl bool

+ Heartbeat time.Duration

+ MaxAckPending int

+ MaxDeliver int

+ MaxWaiting int

+ OptStartSeq uint64

+ OptStartTime *time.Time

+ RateLimit uint64

+ ReplayPolicy ReplayPolicy

+ SampleFrequency string

+ type ConsumerInfo struct

+ AckFloor SequencePair

+ Cluster *ClusterInfo

+ Config ConsumerConfig

+ Created time.Time

+ Delivered SequencePair

+ Name string

+ NumAckPending int

+ NumPending uint64

+ NumRedelivered int

+ NumWaiting int

+ Stream string

+ type ContextOpt struct

+ func Context(ctx context.Context) ContextOpt

+ type DeliverPolicy int

+ const DeliverAllPolicy

+ const DeliverByStartSequencePolicy

+ const DeliverByStartTimePolicy

+ const DeliverLastPolicy

+ const DeliverNewPolicy

+ func (p *DeliverPolicy) UnmarshalJSON(data []byte) error

+ func (p DeliverPolicy) MarshalJSON() ([]byte, error)

+ 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 ErrConsumerSequenceMismatch struct

+ ConsumerSequence uint64

+ LastConsumerSequence uint64

+ StreamResumeSequence uint64

+ func (ecs *ErrConsumerSequenceMismatch) Error() string

+ type ExternalStream struct

+ APIPrefix string

+ DeliverPrefix string

+ type Header map[string][]string

+ func (h Header) Add(key, value string)

+ func (h Header) Del(key string)

+ func (h Header) Get(key string) string

+ func (h Header) Set(key, value string)

+ func (h Header) Values(key string) []string

+ type JSOpt interface

+ func APIPrefix(pre string) JSOpt

+ func PublishAsyncErrHandler(cb MsgErrHandler) JSOpt

+ func PublishAsyncMaxPending(max int) JSOpt

+ type JetStream interface

+ ChanSubscribe func(subj string, ch chan *Msg, opts ...SubOpt) (*Subscription, error)

+ Publish func(subj string, data []byte, opts ...PubOpt) (*PubAck, error)

+ PublishAsync func(subj string, data []byte, opts ...PubOpt) (PubAckFuture, error)

+ PublishAsyncComplete func() <-chan struct{}

+ PublishAsyncPending func() int

+ PublishMsg func(m *Msg, opts ...PubOpt) (*PubAck, error)

+ PublishMsgAsync func(m *Msg, opts ...PubOpt) (PubAckFuture, error)

+ PullSubscribe func(subj, durable string, opts ...SubOpt) (*Subscription, error)

+ QueueSubscribe func(subj, queue string, cb MsgHandler, opts ...SubOpt) (*Subscription, error)

+ QueueSubscribeSync func(subj, queue string, opts ...SubOpt) (*Subscription, error)

+ Subscribe func(subj string, cb MsgHandler, opts ...SubOpt) (*Subscription, error)

+ SubscribeSync func(subj string, opts ...SubOpt) (*Subscription, error)

+ type JetStreamContext interface

+ type JetStreamManager interface

+ AccountInfo func(opts ...JSOpt) (*AccountInfo, error)

+ AddConsumer func(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error)

+ AddStream func(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error)

+ ConsumerInfo func(stream, name string, opts ...JSOpt) (*ConsumerInfo, error)

+ ConsumerNames func(stream string, opts ...JSOpt) <-chan string

+ ConsumersInfo func(stream string, opts ...JSOpt) <-chan *ConsumerInfo

+ DeleteConsumer func(stream, consumer string, opts ...JSOpt) error

+ DeleteMsg func(name string, seq uint64, opts ...JSOpt) error

+ DeleteStream func(name string, opts ...JSOpt) error

+ GetMsg func(name string, seq uint64, opts ...JSOpt) (*RawStreamMsg, error)

+ PurgeStream func(name string, opts ...JSOpt) error

+ StreamInfo func(stream string, opts ...JSOpt) (*StreamInfo, error)

+ StreamNames func(opts ...JSOpt) <-chan string

+ StreamsInfo func(opts ...JSOpt) <-chan *StreamInfo

+ UpdateStream func(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error)

+ type MaxWait time.Duration

type Msg

+ Header Header

+ func NewMsg(subject string) *Msg

+ func (m *Msg) Ack(opts ...AckOpt) error

+ func (m *Msg) AckSync(opts ...AckOpt) error

+ func (m *Msg) InProgress(opts ...AckOpt) error

+ func (m *Msg) Metadata() (*MsgMetadata, error)

+ func (m *Msg) Nak(opts ...AckOpt) error

+ func (m *Msg) RespondMsg(msg *Msg) error

+ func (m *Msg) Term(opts ...AckOpt) error

+ type MsgErrHandler func(JetStream, *Msg, error)

+ type MsgMetadata struct

+ Consumer string

+ NumDelivered uint64

+ NumPending uint64

+ Sequence SequencePair

+ Stream string

+ Timestamp time.Time

type Option

+ func Compression(enabled bool) Option

+ func LameDuckModeHandler(cb ConnHandler) Option

+ func RetryOnFailedConnect(retry bool) Option

type Options

+ Compression bool

+ LameDuckModeHandler ConnHandler

+ RetryOnFailedConnect bool

+ type PeerInfo struct

+ Active time.Duration

+ Current bool

+ Lag uint64

+ Name string

+ Offline bool

+ type Placement struct

+ Cluster string

+ Tags []string

+ type PubAck struct

+ Duplicate bool

+ Sequence uint64

+ Stream string

+ type PubAckFuture interface

+ Err func() <-chan error

+ Msg func() *Msg

+ Ok func() <-chan *PubAck

+ type PubOpt interface

+ func ExpectLastMsgId(id string) PubOpt

+ func ExpectLastSequence(seq uint64) PubOpt

+ func ExpectStream(stream string) PubOpt

+ func MsgId(id string) PubOpt

+ type PullOpt interface

+ type RawStreamMsg struct

+ Data []byte

+ Header Header

+ Sequence uint64

+ Subject string

+ Time time.Time

+ type ReplayPolicy int

+ const ReplayInstantPolicy

+ const ReplayOriginalPolicy

+ func (p *ReplayPolicy) UnmarshalJSON(data []byte) error

+ func (p ReplayPolicy) MarshalJSON() ([]byte, error)

+ 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 SequencePair struct

+ Consumer uint64

+ Stream uint64

+ 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 StreamConfig struct

+ Discard DiscardPolicy

+ Duplicates time.Duration

+ MaxAge time.Duration

+ MaxBytes int64

+ MaxConsumers int

+ MaxMsgSize int32

+ MaxMsgs int64

+ Mirror *StreamSource

+ Name string

+ NoAck bool

+ Placement *Placement

+ Replicas int

+ Retention RetentionPolicy

+ Sources []*StreamSource

+ Storage StorageType

+ Subjects []string

+ Template string

+ type StreamInfo struct

+ Cluster *ClusterInfo

+ Config StreamConfig

+ Created time.Time

+ Mirror *StreamSourceInfo

+ Sources []*StreamSourceInfo

+ State StreamState

+ type StreamSource struct

+ External *ExternalStream

+ FilterSubject string

+ Name string

+ OptStartSeq uint64

+ OptStartTime *time.Time

+ type StreamSourceInfo struct

+ Active time.Duration

+ Lag uint64

+ Name string

+ type StreamState struct

+ Bytes uint64

+ Consumers int

+ FirstSeq uint64

+ FirstTime time.Time

+ LastSeq uint64

+ LastTime time.Time

+ Msgs uint64

+ type SubOpt interface

+ func AckAll() SubOpt

+ func AckExplicit() SubOpt

+ func AckNone() SubOpt

+ func BindStream(name string) SubOpt

+ func DeliverAll() SubOpt

+ func DeliverLast() SubOpt

+ func DeliverNew() SubOpt

+ func Durable(name string) SubOpt

+ func EnableFlowControl() SubOpt

+ func IdleHeartbeat(duration time.Duration) SubOpt

+ func ManualAck() SubOpt

+ func MaxAckPending(n int) SubOpt

+ func MaxDeliver(n int) SubOpt

+ func PullMaxWaiting(n int) SubOpt

+ func RateLimit(n uint64) SubOpt

+ func ReplayOriginal() SubOpt

+ func StartSequence(seq uint64) SubOpt

+ func StartTime(startTime time.Time) SubOpt

type Subscription

+ func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error)

+ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error)

May 14, 2020

Changes in this version

+ const DefaultReconnectJitter

+ const DefaultReconnectJitterTLS

+ var ErrClientIPNotSupported = errors.New("nats: client IP not supported by this server")

+ var ErrDisconnected = errors.New("nats: server is disconnected")

type Conn

+ func (nc *Conn) GetClientIP() (net.IP, error)

+ func (nc *Conn) RTT() (time.Duration, error)

type Option

+ func CustomReconnectDelay(cb ReconnectDelayHandler) Option

+ func ReconnectJitter(jitter, jitterForTLS time.Duration) Option

type Options

+ CustomReconnectDelayCB ReconnectDelayHandler

+ ReconnectJitter time.Duration

+ ReconnectJitterTLS time.Duration

+ type ReconnectDelayHandler func(attempts int) time.Duration

Oct 30, 2019

Changes in this version

+ const AUTHENTICATION_EXPIRED_ERR

+ var ErrAuthExpired = errors.New("nats: authentication expired")

+ var ErrBadQueueName = errors.New("nats: invalid queue name")

type Option

+ func NoCallbacksAfterClientClose() Option

type Options

+ NoCallbacksAfterClientClose bool

Jun 3, 2019

Changes in this version

+ var ErrMsgNoReply = errors.New("nats: message does not have a reply")

+ var ErrMsgNotBound = errors.New("nats: message is not bound to subscription/connection")

type Conn

+ func (nc *Conn) SetDisconnectErrHandler(dcb ConnErrHandler)

+ type ConnErrHandler func(*Conn, error)

type Msg

+ func (m *Msg) Respond(data []byte) error

type Option

+ func DisconnectErrHandler(cb ConnErrHandler) Option

type Options

+ DisconnectedErrCB ConnErrHandler

Feb 21, 2019

Changes in this version

+ var ErrNoDeadlineContext = errors.New("nats: context requires a deadline")

type Conn

+ func (nc *Conn) ConnectedAddr() string

+ func (nc *Conn) FlushWithContext(ctx context.Context) error

type Option

+ func MaxPingsOutstanding(max int) Option

+ func SyncQueueLen(max int) Option

Dec 11, 2018

Changes in this version

+ var ErrClientIDNotSupported = errors.New("nats: client ID not supported by this server")

+ var ErrNkeyAndUser = errors.New("nats: user callback and nkey defined")

+ var ErrNkeyButNoSigCB = errors.New("nats: nkey defined without a signature handler")

+ var ErrNkeysNotSupported = errors.New("nats: nkeys not supported by the server")

+ var ErrNoUserCB = errors.New("nats: user callback not defined")

+ var ErrTokenAlreadySet = errors.New("nats: token and token handler both set")

+ var ErrUserButNoSigCB = errors.New("nats: user callback defined without a signature handler")

+ type AuthTokenHandler func() string

type Conn

+ func (nc *Conn) GetClientID() (uint64, error)

+ func (nc *Conn) NewRespInbox() string

type Option

+ func FlusherTimeout(t time.Duration) Option

+ func Nkey(pubKey string, sigCB SignatureHandler) Option

+ func NkeyOptionFromSeed(seedFile string) (Option, error)

+ func TokenHandler(cb AuthTokenHandler) Option

+ func UserCredentials(userOrChainedFile string, seedFiles ...string) Option

+ func UserJWT(userCB UserJWTHandler, sigCB SignatureHandler) Option

type Options

+ Nkey string

+ SignatureCB SignatureHandler

+ TokenHandler AuthTokenHandler

+ UserJWT UserJWTHandler

+ type SignatureHandler func([]byte) ([]byte, error)

+ type UserJWTHandler func() (string, error)

Aug 29, 2018

Changes in this version

+ const DRAINING_PUBS

+ const DRAINING_SUBS

+ const DefaultDrainTimeout

+ var ErrConnectionDraining = errors.New("nats: connection draining")

+ var ErrConnectionReconnecting = errors.New("nats: connection reconnecting")

+ var ErrDrainTimeout = errors.New("nats: draining connection timed out")

+ var ErrNoEchoNotSupported = errors.New("nats: no echo option not supported by this server")

type Conn

+ func (nc *Conn) Drain() error

+ func (nc *Conn) IsDraining() bool

+ func (nc *Conn) NumSubscriptions() int

type EncodedConn

+ func (c *EncodedConn) Drain() error

type Option

+ func DrainTimeout(t time.Duration) Option

+ func NoEcho() Option

+ func PingInterval(t time.Duration) Option

type Options

+ DrainTimeout time.Duration

+ NoEcho bool

type Subscription

+ func (s *Subscription) Drain() error

Mar 23, 2018

Changes in this version

type Conn

+ func (nc *Conn) Barrier(f func()) error

type Option

+ func ReconnectBufSize(size int) Option

Dec 20, 2017

Changes in this version

+ type CustomDialer interface

+ Dial func(network, address string) (net.Conn, error)

type Option

+ func SetCustomDialer(dialer CustomDialer) Option

type Options

+ CustomDialer CustomDialer

Aug 8, 2017

Changes in this version

+ const AUTHORIZATION_ERR

+ var ErrInvalidContext = errors.New("nats: invalid context")

type Conn

+ func (nc *Conn) RequestWithContext(ctx context.Context, subj string, data []byte) (*Msg, error)

+ func (nc *Conn) SetDiscoveredServersHandler(dscb ConnHandler)

type EncodedConn

+ func (c *EncodedConn) RequestWithContext(ctx context.Context, subject string, v interface{}, vPtr interface{}) error

type Option

+ func DiscoveredServersHandler(cb ConnHandler) Option

+ func UseOldRequestStyle() Option

type Options

+ DiscoveredServersCB ConnHandler

+ FlusherTimeout time.Duration

+ UseOldRequestStyle bool

+ func GetDefaultOptions() Options

type Subscription

+ func (s *Subscription) NextMsgWithContext(ctx context.Context) (*Msg, error)

Oct 26, 2016

Changes in this version

+ const INFO_ARG

+ const OP_I

+ const OP_IN

+ const OP_INF

+ const OP_INFO

+ const OP_INFO_SPC

+ const PERMISSIONS_ERR

type Conn

+ func (nc *Conn) DiscoveredServers() []string

+ func (nc *Conn) IsConnected() bool

+ func (nc *Conn) Servers() []string

type Option

+ func Dialer(dialer *net.Dialer) Option

+ func Token(token string) Option

+ func UserInfo(user, password string) Option

type Options

+ Dialer *net.Dialer

+ Password string

+ Token string

+ User string

May 13, 2016

Changes in this version

+ const AsyncSubscription

+ const ChanSubscription

+ const DefaultReconnectBufSize

+ const DefaultSubPendingBytesLimit

+ const DefaultSubPendingMsgsLimit

+ const NilSubscription

+ const SyncSubscription

+ var ErrInvalidArg = errors.New("nats: invalid argument")

+ var ErrInvalidConnection = errors.New("nats: invalid connection")

+ var ErrInvalidMsg = errors.New("nats: invalid message or message nil")

+ var ErrMaxMessages = errors.New("nats: maximum messages delivered")

+ var ErrMultipleTLSConfigs = errors.New("nats: multiple tls.Configs not allowed")

+ var ErrNoInfoReceived = errors.New("nats: protocol exception, INFO not received")

+ var ErrReconnectBufExceeded = errors.New("nats: outbound buffer limit exceeded")

+ var ErrSyncSubRequired = errors.New("nats: illegal call on an async subscription")

+ var ErrTypeSubscription = errors.New("nats: invalid subscription type")

type Conn

+ func (nc *Conn) AuthRequired() bool

+ func (nc *Conn) ChanQueueSubscribe(subj, group string, ch chan *Msg) (*Subscription, error)

+ func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg) (*Subscription, error)

+ func (nc *Conn) QueueSubscribeSyncWithChan(subj, queue string, ch chan *Msg) (*Subscription, error)

+ func (nc *Conn) SetClosedHandler(cb ConnHandler)

+ func (nc *Conn) SetDisconnectHandler(dcb ConnHandler)

+ func (nc *Conn) SetErrorHandler(cb ErrHandler)

+ func (nc *Conn) SetReconnectHandler(rcb ConnHandler)

+ func (nc *Conn) TLSRequired() bool

+ type Option func(*Options) error

+ func ClientCert(certFile, keyFile string) Option

+ func ClosedHandler(cb ConnHandler) Option

+ func DisconnectHandler(cb ConnHandler) Option

+ func DontRandomize() Option

+ func ErrorHandler(cb ErrHandler) Option

+ func MaxReconnects(max int) Option

+ func Name(name string) Option

+ func NoReconnect() Option

+ func ReconnectHandler(cb ConnHandler) Option

+ func ReconnectWait(t time.Duration) Option

+ func RootCAs(file ...string) Option

+ func Secure(tls ...*tls.Config) Option

+ func Timeout(t time.Duration) Option

type Options

+ ReconnectBufSize int

type Subscription

+ func (s *Subscription) ClearMaxPending() error

+ func (s *Subscription) Delivered() (int64, error)

+ func (s *Subscription) Dropped() (int, error)

+ func (s *Subscription) MaxPending() (int, int, error)

+ func (s *Subscription) Pending() (int, int, error)

+ func (s *Subscription) PendingLimits() (int, int, error)

+ func (s *Subscription) SetPendingLimits(msgLimit, bytesLimit int) error

+ func (s *Subscription) Type() SubscriptionType

+ type SubscriptionType int

Dec 9, 2015

Changes in this version

+ var ErrSecureConnFailed = errors.New("nats: Secure Connection failed")

type Options

+ TLSConfig *tls.Config

Sep 18, 2015

Changes in this version

+ const CONNECTING

+ var ErrBadSubject = errors.New("nats: Invalid Subject")

+ var ErrMaxPayload = errors.New("nats: Maximum Payload Exceeded")

type Conn

+ func (nc *Conn) Buffered() (int, error)

+ func (nc *Conn) MaxPayload() int64

type Subscription

+ func (s *Subscription) QueuedMsgs() (int, error)

Jun 19, 2015

Changes in this version

+ const DefaultMaxChanLen

+ const LangString

+ const RequestChanLen

type Options

+ SubChanLen int

Nov 28, 2014

Changes in this version

+ const DEFAULT_ENCODER

+ const DefaultMaxPingOut

+ const DefaultPingInterval

+ const GOB_ENCODER

+ const JSON_ENCODER

+ const STALE_CONNECTION

+ var ErrStaleConnection = errors.New("nats: " + STALE_CONNECTION)

type Conn

+ func (nc *Conn) ConnectedServerId() string

type Options

+ MaxPingsOut int

+ PingInterval time.Duration

Oct 22, 2014

Changes in this version

+ const CLOSED

+ const CONNECTED

+ const CONNECT_ARG

+ const DISCONNECTED

+ const DefaultMaxReconnect

+ const DefaultPort

+ const DefaultReconnectWait

+ const DefaultTimeout

+ const DefaultURL

+ const InboxPrefix

+ const MAX_CONTROL_LINE_SIZE

+ const MINUS_ERR_ARG

+ const MSG_ARG

+ const MSG_END

+ const MSG_PAYLOAD

+ const OP_C

+ const OP_CO

+ const OP_CON

+ const OP_CONN

+ const OP_CONNE

+ const OP_CONNEC

+ const OP_CONNECT

+ const OP_M

+ const OP_MINUS

+ const OP_MINUS_E

+ const OP_MINUS_ER

+ const OP_MINUS_ERR

+ const OP_MINUS_ERR_SPC

+ const OP_MS

+ const OP_MSG

+ const OP_MSG_SPC

+ const OP_P

+ const OP_PI

+ const OP_PIN

+ const OP_PING

+ const OP_PLUS

+ const OP_PLUS_O

+ const OP_PLUS_OK

+ const OP_PO

+ const OP_PON

+ const OP_PONG

+ const OP_START

+ const RECONNECTING

+ const Version

+ var DefaultOptions = Options

+ var ErrAuthorization = errors.New("nats: Authorization Failed")

+ var ErrBadSubscription = errors.New("nats: Invalid Subscription")

+ var ErrBadTimeout = errors.New("nats: Timeout Invalid")

+ var ErrChanArg = errors.New("nats: Argument needs to be a channel type")

+ var ErrConnectionClosed = errors.New("nats: Connection Closed")

+ var ErrJsonParse = errors.New("nats: Connect message, json parse err")

+ var ErrNoServers = errors.New("nats: No servers available for connection")

+ var ErrSecureConnRequired = errors.New("nats: Secure connection required")

+ var ErrSecureConnWanted = errors.New("nats: Secure connection not available")

+ var ErrSlowConsumer = errors.New("nats: Slow Consumer, messages dropped")

+ var ErrTimeout = errors.New("nats: Timeout")

+ func NewInbox() string

+ func RegisterEncoder(encType string, enc Encoder)

+ type Conn struct

+ Opts Options

+ func Connect(url string) (*Conn, error)

+ func SecureConnect(url string) (*Conn, error)

+ func (nc *Conn) Close()

+ func (nc *Conn) ConnectedUrl() string

+ func (nc *Conn) Flush() error

+ func (nc *Conn) FlushTimeout(timeout time.Duration) (err error)

+ func (nc *Conn) IsClosed() bool

+ func (nc *Conn) IsReconnecting() bool

+ func (nc *Conn) LastError() error

+ func (nc *Conn) Publish(subj string, data []byte) error

+ func (nc *Conn) PublishMsg(m *Msg) error

+ func (nc *Conn) PublishRequest(subj, reply string, data []byte) error

+ func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler) (*Subscription, error)

+ func (nc *Conn) QueueSubscribeSync(subj, queue string) (*Subscription, error)

+ func (nc *Conn) Request(subj string, data []byte, timeout time.Duration) (m *Msg, err error)

+ func (nc *Conn) Stats() Statistics

+ func (nc *Conn) Status() Status

+ func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error)

+ func (nc *Conn) SubscribeSync(subj string) (*Subscription, error)

+ type ConnHandler func(*Conn)

+ type DefaultEncoder struct

+ func (je *DefaultEncoder) Decode(subject string, data []byte, vPtr interface{}) error

+ func (je *DefaultEncoder) Encode(subject string, v interface{}) ([]byte, error)

+ type EncodedConn struct

+ Conn *Conn

+ Enc Encoder

+ func NewEncodedConn(c *Conn, encType string) (*EncodedConn, error)

+ func (c *EncodedConn) BindRecvChan(subject string, channel interface{}) (*Subscription, error)

+ func (c *EncodedConn) BindRecvQueueChan(subject, queue string, channel interface{}) (*Subscription, error)

+ func (c *EncodedConn) BindSendChan(subject string, channel interface{}) error

+ func (c *EncodedConn) Close()

+ func (c *EncodedConn) Flush() error

+ func (c *EncodedConn) FlushTimeout(timeout time.Duration) (err error)

+ func (c *EncodedConn) LastError() error

+ func (c *EncodedConn) Publish(subject string, v interface{}) error

+ func (c *EncodedConn) PublishRequest(subject, reply string, v interface{}) error

+ func (c *EncodedConn) QueueSubscribe(subject, queue string, cb Handler) (*Subscription, error)

+ func (c *EncodedConn) Request(subject string, v interface{}, vPtr interface{}, timeout time.Duration) error

+ func (c *EncodedConn) Subscribe(subject string, cb Handler) (*Subscription, error)

+ type Encoder interface

+ Decode func(subject string, data []byte, vPtr interface{}) error

+ Encode func(subject string, v interface{}) ([]byte, error)

+ func EncoderForType(encType string) Encoder

+ type ErrHandler func(*Conn, *Subscription, error)

+ type GobEncoder struct

+ func (ge *GobEncoder) Decode(subject string, data []byte, vPtr interface{}) (err error)

+ func (ge *GobEncoder) Encode(subject string, v interface{}) ([]byte, error)

+ type Handler interface

+ type JsonEncoder struct

+ func (je *JsonEncoder) Decode(subject string, data []byte, vPtr interface{}) (err error)

+ func (je *JsonEncoder) Encode(subject string, v interface{}) ([]byte, error)

+ type Msg struct

+ Data []byte

+ Reply string

+ Sub *Subscription

+ Subject string

+ type MsgHandler func(msg *Msg)

+ type Options struct

+ AllowReconnect bool

+ AsyncErrorCB ErrHandler

+ ClosedCB ConnHandler

+ DisconnectedCB ConnHandler

+ MaxReconnect int

+ Name string

+ NoRandomize bool

+ Pedantic bool

+ ReconnectWait time.Duration

+ ReconnectedCB ConnHandler

+ Secure bool

+ Servers []string

+ Timeout time.Duration

+ Url string

+ Verbose bool

+ func (o Options) Connect() (*Conn, error)

+ type Statistics struct

+ InBytes uint64

+ InMsgs uint64

+ OutBytes uint64

+ OutMsgs uint64

+ Reconnects uint64

+ type Status int

+ type Subscription struct

+ Queue string

+ Subject string

+ func (s *Subscription) AutoUnsubscribe(max int) error

+ func (s *Subscription) IsValid() bool

+ func (s *Subscription) NextMsg(timeout time.Duration) (msg *Msg, err error)

+ func (s *Subscription) Unsubscribe() error