stores

Apr 26, 2019

Changes in this version

type RaftSubStore

+ func (gss *RaftSubStore) Close() error

+ func (gss *RaftSubStore) DeleteSub(subid uint64) error

+ func (gss *RaftSubStore) Flush() error

+ func (ss *RaftSubStore) CreateSub(sub *spb.SubState) error

Feb 8, 2019

Changes in this version

+ const CryptoCipherAES

+ const CryptoCipherAutoSelect

+ const CryptoCipherChaChaPoly

+ const CryptoCodeAES

+ const CryptoCodeChaCha

+ const CryptoStoreEnvKeyName

+ var ErrCipherNotSupported = errors.New("encryption cipher not supported")

+ var ErrCryptoStoreRequiresKey = errors.New("encryption key required")

+ type CryptoMsgStore struct

+ func (cms *CryptoMsgStore) FirstMsg() (*pb.MsgProto, error)

+ func (cms *CryptoMsgStore) LastMsg() (*pb.MsgProto, error)

+ func (cms *CryptoMsgStore) Lookup(seq uint64) (*pb.MsgProto, error)

+ func (cms *CryptoMsgStore) Store(msg *pb.MsgProto) (uint64, error)

+ type CryptoStore struct

+ func NewCryptoStore(s Store, encryptionCipher string, encryptionKey []byte) (*CryptoStore, error)

+ func (cs *CryptoStore) CreateChannel(channel string) (*Channel, error)

+ func (cs *CryptoStore) Recover() (*RecoveredState, error)

+ type EDStore struct

+ func NewEDStore(encryptionCipher string, encryptionKey []byte, idx uint64) (*EDStore, error)

+ func (s *EDStore) Decrypt(dst []byte, cipherText []byte) ([]byte, error)

+ func (s *EDStore) Encrypt(pbuf *[]byte, data []byte) ([]byte, error)

+ func (s *EDStore) EncryptionOffset() int

Jun 19, 2018

Changes in this version

type FileStoreOption

+ func TruncateUnexpectedEOF(truncate bool) FileStoreOption

type FileStoreOptions

+ TruncateUnexpectedEOF bool

Mar 16, 2018

Changes in this version

+ const TypeRaft

+ var ErrNotFound = errors.New("not found")

type ChannelLimits

+ MaxInactivity time.Duration

type FileMsgStore

+ func (ms *FileMsgStore) Empty() error

type FileStore

+ func (fs *FileStore) DeleteChannel(channel string) error

+ func (gs *FileStore) GetChannelLimits(channel string) *ChannelLimits

type MemoryMsgStore

+ func (ms *MemoryMsgStore) Empty() error

type MemoryStore

+ func (gs *MemoryStore) DeleteChannel(channel string) error

+ func (gs *MemoryStore) GetChannelLimits(channel string) *ChannelLimits

type MsgStore

+ Empty func() error

+ type RaftStore struct

+ func NewRaftStore(s Store) *RaftStore

+ func (s *RaftStore) AddClient(clientID, hbInbox string) (*Client, error)

+ func (s *RaftStore) CreateChannel(channel string) (*Channel, error)

+ func (s *RaftStore) DeleteClient(clientID string) error

+ func (s *RaftStore) Name() string

+ func (s *RaftStore) Recover() (*RecoveredState, error)

+ type RaftSubStore struct

+ func (ss *RaftSubStore) AckSeqPending(subid, seqno uint64) error

+ func (ss *RaftSubStore) AddSeqPending(subid, seqno uint64) error

+ func (ss *RaftSubStore) UpdateSub(*spb.SubState) error

type SQLMsgStore

+ func (ms *SQLMsgStore) Empty() error

type SQLStore

+ func (gs *SQLStore) GetChannelLimits(channel string) *ChannelLimits

+ func (s *SQLStore) DeleteChannel(channel string) error

type Store

+ DeleteChannel func(channel string) error

+ GetChannelLimits func(name string) *ChannelLimits

Dec 19, 2017

Changes in this version

+ const TypeSQL

+ type SQLMsgStore struct

+ func (gms *SQLMsgStore) FirstAndLastSequence() (uint64, uint64, error)

+ func (gms *SQLMsgStore) FirstSequence() (uint64, error)

+ func (gms *SQLMsgStore) LastSequence() (uint64, error)

+ func (gms *SQLMsgStore) State() (numMessages int, byteSize uint64, err error)

+ func (ms *SQLMsgStore) Close() error

+ func (ms *SQLMsgStore) FirstMsg() (*pb.MsgProto, error)

+ func (ms *SQLMsgStore) Flush() error

+ func (ms *SQLMsgStore) GetSequenceFromTimestamp(timestamp int64) (uint64, error)

+ func (ms *SQLMsgStore) LastMsg() (*pb.MsgProto, error)

+ func (ms *SQLMsgStore) Lookup(seq uint64) (*pb.MsgProto, error)

+ func (ms *SQLMsgStore) Store(data []byte) (uint64, error)

+ type SQLStore struct

+ func NewSQLStore(log logger.Logger, driver, source string, limits *StoreLimits, ...) (*SQLStore, error)

+ func (gs *SQLStore) Name() string

+ func (gs *SQLStore) SetLimits(limits *StoreLimits) error

+ func (s *SQLStore) AddClient(clientID, hbInbox string) (*Client, error)

+ func (s *SQLStore) Close() error

+ func (s *SQLStore) CreateChannel(channel string) (*Channel, error)

+ func (s *SQLStore) DeleteClient(clientID string) error

+ func (s *SQLStore) GetExclusiveLock() (bool, error)

+ func (s *SQLStore) Init(info *spb.ServerInfo) error

+ func (s *SQLStore) Recover() (*RecoveredState, error)

+ type SQLStoreOption func(*SQLStoreOptions) error

+ func SQLAllOptions(opts *SQLStoreOptions) SQLStoreOption

+ func SQLMaxOpenConns(max int) SQLStoreOption

+ func SQLNoCaching(noCaching bool) SQLStoreOption

+ type SQLStoreOptions struct

+ Driver string

+ MaxOpenConns int

+ NoCaching bool

+ Source string

+ func DefaultSQLStoreOptions() *SQLStoreOptions

+ type SQLSubStore struct

+ func (ss *SQLSubStore) AckSeqPending(subid, seqno uint64) error

+ func (ss *SQLSubStore) AddSeqPending(subid, seqno uint64) error

+ func (ss *SQLSubStore) Close() error

+ func (ss *SQLSubStore) CreateSub(sub *spb.SubState) error

+ func (ss *SQLSubStore) DeleteSub(subid uint64) error

+ func (ss *SQLSubStore) Flush() error

+ func (ss *SQLSubStore) UpdateSub(sub *spb.SubState) error

Oct 25, 2017

Changes in this version

+ var ErrAlreadyExists = errors.New("already exists")

+ func FileStoreTestResetBackgroundTaskInterval()

+ func FileStoreTestSetBackgroundTaskInterval(wait time.Duration)

+ type Channel struct

+ Msgs MsgStore

+ Subs SubStore

+ type RecoveredChannel struct

+ Channel *Channel

+ Subscriptions []*RecoveredSubscription

type RecoveredState

+ Channels map[string]*RecoveredChannel

+ type RecoveredSubscription struct

+ Pending PendingAcks

+ Sub *spb.SubState

type StoreLimits

+ func (sl *StoreLimits) ClonePerChannelMap() map[string]*ChannelLimits

Jun 27, 2017

Changes in this version

type DelegateStore

+ func (ms *DelegateStore) GetChannels() map[string]*ChannelStore

+ func (ms *DelegateStore) GetChannelsCount() int

type FileStore

+ func (gs *FileStore) GetChannels() map[string]*ChannelStore

+ func (gs *FileStore) GetChannelsCount() int

type FileStoreOption

+ func ParallelRecovery(count int) FileStoreOption

type FileStoreOptions

+ ParallelRecovery int

type MemoryStore

+ func (gs *MemoryStore) GetChannels() map[string]*ChannelStore

+ func (gs *MemoryStore) GetChannelsCount() int

type Store

+ GetChannels func() map[string]*ChannelStore

+ GetChannelsCount func() int

type StoreLimits

+ func (sl *StoreLimits) Clone() *StoreLimits

+ func (sl *StoreLimits) Print() []string

Apr 10, 2017

Changes in this version

+ var ErrNotSupported = errors.New("not supported")

+ type DelegateStore struct

+ S Store

+ func (ms *DelegateStore) AddClient(clientID, hbInbox string, userData interface{}) (*Client, bool, error)

+ func (ms *DelegateStore) Close() error

+ func (ms *DelegateStore) CreateChannel(channel string, userData interface{}) (*ChannelStore, bool, error)

+ func (ms *DelegateStore) DeleteClient(clientID string) *Client

+ func (ms *DelegateStore) GetClient(clientID string) *Client

+ func (ms *DelegateStore) GetClients() map[string]*Client

+ func (ms *DelegateStore) GetClientsCount() int

+ func (ms *DelegateStore) GetExclusiveLock() (bool, error)

+ func (ms *DelegateStore) HasChannel() bool

+ func (ms *DelegateStore) Init(si *spb.ServerInfo) error

+ func (ms *DelegateStore) LookupChannel(channel string) *ChannelStore

+ func (ms *DelegateStore) MsgsState(channel string) (numMessages int, byteSize uint64, err error)

+ func (ms *DelegateStore) Name() string

+ func (ms *DelegateStore) Recover() (*RecoveredState, error)

+ func (ms *DelegateStore) SetLimits(limits *StoreLimits) error

type FileStore

+ func (fs *FileStore) GetExclusiveLock() (bool, error)

+ func (fs *FileStore) Recover() (*RecoveredState, error)

type FileStoreOption

+ func FileDescriptorsLimit(limit int64) FileStoreOption

type FileStoreOptions

+ FileDescriptorsLimit int64

type MemoryStore

+ func (gs *MemoryStore) GetExclusiveLock() (bool, error)

+ func (gs *MemoryStore) Recover() (*RecoveredState, error)

type Store

+ GetExclusiveLock func() (bool, error)

+ Recover func() (*RecoveredState, error)

Dec 12, 2016

Changes in this version

type FileStoreOption

+ func SliceConfig(maxMsgs int, maxBytes int64, maxAge time.Duration, script string) FileStoreOption

type FileStoreOptions

+ SliceArchiveScript string

+ SliceMaxAge time.Duration

+ SliceMaxBytes int64

+ SliceMaxMsgs int

Oct 19, 2016

Changes in this version

+ var DefaultStoreLimits = StoreLimits

type FileStore

+ func (gs *FileStore) SetLimits(limits *StoreLimits) error

type FileStoreOption

+ func CacheMsgs(cache bool) FileStoreOption

type FileStoreOptions

+ CacheMsgs bool

type MemoryStore

+ func (gs *MemoryStore) SetLimits(limits *StoreLimits) error

+ type MsgStoreLimits struct

+ MaxAge time.Duration

+ MaxBytes int64

+ MaxMsgs int

type Store

+ SetLimits func(limits *StoreLimits) error

+ type StoreLimits struct

+ MaxChannels int

+ PerChannel map[string]*ChannelLimits

+ func (sl *StoreLimits) AddPerChannel(name string, cl *ChannelLimits)

+ func (sl *StoreLimits) Build() error

+ type SubStoreLimits struct

+ MaxSubscriptions int

Aug 8, 2016

Changes in this version

type FileStoreOption

+ func DoSync(enableFileSync bool) FileStoreOption

type FileStoreOptions

+ DoSync bool

Jul 8, 2016

Changes in this version

type FileMsgStore

+ func (ms *FileMsgStore) Flush() error

type FileStoreOption

+ func BufferSize(size int) FileStoreOption

+ func CRCPolynomial(polynomial int64) FileStoreOption

+ func DoCRC(enableCRC bool) FileStoreOption

type FileStoreOptions

+ BufferSize int

+ CRCPolynomial int64

+ DoCRC bool

type FileSubStore

+ func (ss *FileSubStore) Flush() error

type MemoryMsgStore

+ func (gms *MemoryMsgStore) Flush() error

type MemorySubStore

+ func (gss *MemorySubStore) Flush() error

type MsgStore

+ Flush func() error

type SubStore

+ Flush func() error

Jun 24, 2016

Changes in this version

+ const AllChannels

+ const TypeFile

+ const TypeMemory

+ var DefaultChannelLimits = ChannelLimits

+ var DefaultFileStoreOptions = FileStoreOptions

+ var ErrTooManyChannels = errors.New("too many channels")

+ var ErrTooManySubs = errors.New("too many subscriptions per channel")

+ func NewFileStore(rootDir string, limits *ChannelLimits, options ...FileStoreOption) (*FileStore, *RecoveredState, error)

+ func Noticef(format string, v ...interface{})

+ type ChannelLimits struct

+ MaxChannels int

+ MaxMsgAge time.Duration

+ MaxMsgBytes uint64

+ MaxNumMsgs int

+ MaxSubs int

+ type ChannelStore struct

+ Msgs MsgStore

+ Subs SubStore

+ UserData interface{}

+ type Client struct

+ ClientID string

+ HbInbox string

+ UserData interface{}

+ type FileMsgStore struct

+ func (gms *FileMsgStore) FirstAndLastSequence() (uint64, uint64)

+ func (gms *FileMsgStore) FirstMsg() *pb.MsgProto

+ func (gms *FileMsgStore) FirstSequence() uint64

+ func (gms *FileMsgStore) GetSequenceFromTimestamp(timestamp int64) uint64

+ func (gms *FileMsgStore) LastMsg() *pb.MsgProto

+ func (gms *FileMsgStore) LastSequence() uint64

+ func (gms *FileMsgStore) Lookup(seq uint64) *pb.MsgProto

+ func (gms *FileMsgStore) State() (numMessages int, byteSize uint64, err error)

+ func (ms *FileMsgStore) Close() error

+ func (ms *FileMsgStore) Store(reply string, data []byte) (*pb.MsgProto, error)

+ type FileStore struct

+ func (fs *FileStore) AddClient(clientID, hbInbox string, userData interface{}) (*Client, bool, error)

+ func (fs *FileStore) Close() error

+ func (fs *FileStore) CreateChannel(channel string, userData interface{}) (*ChannelStore, bool, error)

+ func (fs *FileStore) DeleteClient(clientID string) *Client

+ func (fs *FileStore) Init(info *spb.ServerInfo) error

+ func (gs *FileStore) GetClient(clientID string) *Client

+ func (gs *FileStore) GetClients() map[string]*Client

+ func (gs *FileStore) GetClientsCount() int

+ func (gs *FileStore) HasChannel() bool

+ func (gs *FileStore) LookupChannel(channel string) *ChannelStore

+ func (gs *FileStore) MsgsState(channel string) (numMessages int, byteSize uint64, err error)

+ func (gs *FileStore) Name() string

+ func (gs *FileStore) SetChannelLimits(limits ChannelLimits)

+ type FileStoreOption func(*FileStoreOptions) error

+ func AllOptions(opts *FileStoreOptions) FileStoreOption

+ func CompactEnabled(enabled bool) FileStoreOption

+ func CompactFragmentation(fragmentation int) FileStoreOption

+ func CompactInterval(seconds int) FileStoreOption

+ func CompactMinFileSize(fileSize int64) FileStoreOption

+ type FileStoreOptions struct

+ CompactEnabled bool

+ CompactFragmentation int

+ CompactInterval int

+ CompactMinFileSize int64

+ type FileSubStore struct

+ func (ss *FileSubStore) AckSeqPending(subid, seqno uint64) error

+ func (ss *FileSubStore) AddSeqPending(subid, seqno uint64) error

+ func (ss *FileSubStore) Close() error

+ func (ss *FileSubStore) CreateSub(sub *spb.SubState) error

+ func (ss *FileSubStore) DeleteSub(subid uint64)

+ func (ss *FileSubStore) UpdateSub(sub *spb.SubState) error

+ type MemoryMsgStore struct

+ func (gms *MemoryMsgStore) Close() error

+ func (gms *MemoryMsgStore) FirstAndLastSequence() (uint64, uint64)

+ func (gms *MemoryMsgStore) FirstMsg() *pb.MsgProto

+ func (gms *MemoryMsgStore) FirstSequence() uint64

+ func (gms *MemoryMsgStore) GetSequenceFromTimestamp(timestamp int64) uint64

+ func (gms *MemoryMsgStore) LastMsg() *pb.MsgProto

+ func (gms *MemoryMsgStore) LastSequence() uint64

+ func (gms *MemoryMsgStore) Lookup(seq uint64) *pb.MsgProto

+ func (gms *MemoryMsgStore) State() (numMessages int, byteSize uint64, err error)

+ func (ms *MemoryMsgStore) Store(reply string, data []byte) (*pb.MsgProto, error)

+ type MemoryStore struct

+ func NewMemoryStore(limits *ChannelLimits) (*MemoryStore, error)

+ func (gs *MemoryStore) AddClient(clientID, hbInbox string, userData interface{}) (*Client, bool, error)

+ func (gs *MemoryStore) Close() error

+ func (gs *MemoryStore) DeleteClient(clientID string) *Client

+ func (gs *MemoryStore) GetClient(clientID string) *Client

+ func (gs *MemoryStore) GetClients() map[string]*Client

+ func (gs *MemoryStore) GetClientsCount() int

+ func (gs *MemoryStore) HasChannel() bool

+ func (gs *MemoryStore) Init(info *spb.ServerInfo) error

+ func (gs *MemoryStore) LookupChannel(channel string) *ChannelStore

+ func (gs *MemoryStore) MsgsState(channel string) (numMessages int, byteSize uint64, err error)

+ func (gs *MemoryStore) Name() string

+ func (gs *MemoryStore) SetChannelLimits(limits ChannelLimits)

+ func (ms *MemoryStore) CreateChannel(channel string, userData interface{}) (*ChannelStore, bool, error)

+ type MemorySubStore struct

+ func (*MemorySubStore) AckSeqPending(subid, seqno uint64) error

+ func (*MemorySubStore) AddSeqPending(subid, seqno uint64) error

+ func (gss *MemorySubStore) Close() error

+ func (gss *MemorySubStore) CreateSub(sub *spb.SubState) error

+ func (gss *MemorySubStore) DeleteSub(subid uint64)

+ func (gss *MemorySubStore) UpdateSub(sub *spb.SubState) error

+ type MsgStore interface

+ Close func() error

+ FirstAndLastSequence func() (uint64, uint64)

+ FirstMsg func() *pb.MsgProto

+ FirstSequence func() uint64

+ GetSequenceFromTimestamp func(timestamp int64) uint64

+ LastMsg func() *pb.MsgProto

+ LastSequence func() uint64

+ Lookup func(seq uint64) *pb.MsgProto

+ State func() (numMessages int, byteSize uint64, err error)

+ Store func(reply string, data []byte) (*pb.MsgProto, error)

+ type PendingAcks map[uint64]*pb.MsgProto

+ type RecoveredState struct

+ Clients []*Client

+ Info *spb.ServerInfo

+ Subs RecoveredSubscriptions

+ type RecoveredSubState struct

+ Pending PendingAcks

+ Sub *spb.SubState

+ type RecoveredSubscriptions map[string][]*RecoveredSubState

+ type Store interface

+ AddClient func(clientID, hbInbox string, userData interface{}) (*Client, bool, error)

+ Close func() error

+ CreateChannel func(channel string, userData interface{}) (*ChannelStore, bool, error)

+ DeleteClient func(clientID string) *Client

+ GetClient func(clientID string) *Client

+ GetClients func() map[string]*Client

+ GetClientsCount func() int

+ HasChannel func() bool

+ Init func(info *spb.ServerInfo) error

+ LookupChannel func(channel string) *ChannelStore

+ MsgsState func(channel string) (numMessages int, byteSize uint64, err error)

+ Name func() string

+ SetChannelLimits func(limits ChannelLimits)

+ type SubStore interface

+ AckSeqPending func(subid, seqno uint64) error

+ AddSeqPending func(subid, seqno uint64) error

+ Close func() error

+ CreateSub func(*spb.SubState) error

+ DeleteSub func(subid uint64)

+ UpdateSub func(*spb.SubState) error