msq

Documentation

Index

• type Config
  • func DefaultMsgConfig() *Config
  • func (c *Config) Check() error
• type DeadSessionID
  • func NewDeadSessionID(config *Config) *DeadSessionID
  • func (ds *DeadSessionID) Clean()
  • func (ds *DeadSessionID) Exists(sid string) bool
  • func (ds *DeadSessionID) Push(sid string)
• type MemControl
  • func NewMemControl(config *Config) *MemControl
  • func (mc *MemControl) Prefetch(byteSize uint64, timeout time.Duration) (bool, time.Duration)
  • func (mc *MemControl) Release(byteSize uint64)
• type Message
  • func NewMessage(msg []byte) *Message
  • func (m *Message) ByteSize() uint64
• type Session
• type SessionIDItem
  • func NewSessionIDItem(id string, t int64) *SessionIDItem
• type SessionIDPQ
  • func (pq *SessionIDPQ) Len() int
  • func (pq *SessionIDPQ) Less(i, j int) bool
  • func (pq *SessionIDPQ) Pop() interface{}
  • func (pq *SessionIDPQ) Push(x interface{})
  • func (pq *SessionIDPQ) Swap(i, j int)
• type SessionManager
  • func NewSessionManager(config *Config) *SessionManager
  • func (s *SessionManager) GetOrCreateSession(sid string, refresh bool) (*SessionQueue, *transerr.TransError)
  • func (s *SessionManager) GetSession(sid string, refresh bool) (*SessionQueue, *transerr.TransError)
  • func (s *SessionManager) Peek(sid, topic string) (*Message, *transerr.TransError)
  • func (s *SessionManager) Pop(sid, topic string, timeout time.Duration) (*Message, *transerr.TransError)
  • func (s *SessionManager) Push(sid, topic string, message *Message, timeout time.Duration) *transerr.TransError
  • func (s *SessionManager) ReleaseSession(sid string)
  • func (s *SessionManager) ReleaseTopic(sid, topic string)
  • func (s *SessionManager) StartCleanLoop(stopCh <-chan struct{})
• type SessionQueue
  • func NewSessionQueue(config *Config) *SessionQueue
  • func (s *SessionQueue) Peek(topic string) (*Message, *transerr.TransError)
  • func (s *SessionQueue) Pop(topic string, timeout time.Duration) (*Message, *transerr.TransError)
  • func (s *SessionQueue) Push(topic string, message *Message, timeout time.Duration) *transerr.TransError
  • func (s *SessionQueue) ReleaseSession() uint64
  • func (s *SessionQueue) ReleaseTopic(topic string) uint64
• type Topic
  • func NewTopicQueue(topic string, topicQueueCapacity int) *Topic
  • func (t *Topic) Len() int
  • func (t *Topic) Pop() *Message
  • func (t *Topic) Push(message *Message)

Types

type Config

type Config struct {
	TotalByteSizeLimit      uint64 `yaml:"totalByteSizeLimit,omitempty"`
	PerSessionByteSizeLimit uint64 `yaml:"perSessionByteSizeLimit,omitempty"`
	TopicQueueCapacity      int    `yaml:"topicQueueCapacity,omitempty"`

	DeadSessionIDExpireSeconds int64 `yaml:"deadSessionIDExpireSeconds,omitempty"`
	SessionExpireSeconds       int64 `yaml:"sessionExpireSeconds,omitempty"`
	NormalizeActiveSeconds     int64 `yaml:"normalizeActiveSeconds,omitempty"`

	CleanIntervalSeconds int64 `yaml:"cleanIntervalSeconds,omitempty"`
}

func DefaultMsgConfig

func DefaultMsgConfig() *Config

func (*Config) Check

func (c *Config) Check() error

type DeadSessionID

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

func NewDeadSessionID

func NewDeadSessionID(config *Config) *DeadSessionID

func (*DeadSessionID) Clean

func (ds *DeadSessionID) Clean()

func (*DeadSessionID) Exists

func (ds *DeadSessionID) Exists(sid string) bool

func (*DeadSessionID) Push

func (ds *DeadSessionID) Push(sid string)

type MemControl

type MemControl struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewMemControl

func NewMemControl(config *Config) *MemControl

func (*MemControl) Prefetch

func (mc *MemControl) Prefetch(byteSize uint64, timeout time.Duration) (bool, time.Duration)

func (*MemControl) Release

func (mc *MemControl) Release(byteSize uint64)

type Message

type Message struct {
	Content []byte
}

func NewMessage

func NewMessage(msg []byte) *Message

func (*Message) ByteSize

func (m *Message) ByteSize() uint64

type Session

type Session struct {
	Queue      *SessionQueue
	ActiveMark *SessionIDItem
}

type SessionIDItem

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

func NewSessionIDItem

func NewSessionIDItem(id string, t int64) *SessionIDItem

type SessionIDPQ

type SessionIDPQ []*SessionIDItem

func (*SessionIDPQ) Len

func (pq *SessionIDPQ) Len() int

func (*SessionIDPQ) Less

func (pq *SessionIDPQ) Less(i, j int) bool

func (*SessionIDPQ) Pop

func (pq *SessionIDPQ) Pop() interface{}

func (*SessionIDPQ) Push

func (pq *SessionIDPQ) Push(x interface{})

func (*SessionIDPQ) Swap

func (pq *SessionIDPQ) Swap(i, j int)

type SessionManager

type SessionManager struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewSessionManager

func NewSessionManager(config *Config) *SessionManager

func (*SessionManager) GetOrCreateSession

func (s *SessionManager) GetOrCreateSession(sid string, refresh bool) (*SessionQueue, *transerr.TransError)

func (*SessionManager) GetSession

func (s *SessionManager) GetSession(sid string, refresh bool) (*SessionQueue, *transerr.TransError)

func (*SessionManager) Peek

func (s *SessionManager) Peek(sid, topic string) (*Message, *transerr.TransError)

func (*SessionManager) Pop

func (s *SessionManager) Pop(sid, topic string, timeout time.Duration) (*Message, *transerr.TransError)

func (*SessionManager) Push

func (s *SessionManager) Push(sid, topic string, message *Message, timeout time.Duration) *transerr.TransError

func (*SessionManager) ReleaseSession

func (s *SessionManager) ReleaseSession(sid string)

func (*SessionManager) ReleaseTopic

func (s *SessionManager) ReleaseTopic(sid, topic string)

func (*SessionManager) StartCleanLoop

func (s *SessionManager) StartCleanLoop(stopCh <-chan struct{})

type SessionQueue

type SessionQueue struct {
	ByteSizeLimit uint64
	ByteSize      uint64
	// contains filtered or unexported fields
}

func NewSessionQueue

func NewSessionQueue(config *Config) *SessionQueue

func (*SessionQueue) Peek

func (s *SessionQueue) Peek(topic string) (*Message, *transerr.TransError)

func (*SessionQueue) Pop

func (s *SessionQueue) Pop(topic string, timeout time.Duration) (*Message, *transerr.TransError)

func (*SessionQueue) Push

func (s *SessionQueue) Push(topic string, message *Message, timeout time.Duration) *transerr.TransError

func (*SessionQueue) ReleaseSession

func (s *SessionQueue) ReleaseSession() uint64

func (*SessionQueue) ReleaseTopic

func (s *SessionQueue) ReleaseTopic(topic string) uint64

type Topic

type Topic struct {
	ByteSize uint64
	// contains filtered or unexported fields
}

func NewTopicQueue

func NewTopicQueue(topic string, topicQueueCapacity int) *Topic

func (*Topic) Len

func (t *Topic) Len() int

func (*Topic) Pop

func (t *Topic) Pop() *Message

func (*Topic) Push

func (t *Topic) Push(message *Message)