message

Documentation

Index

• Constants
• Variables
• type Awaiter
• type Counter
• type Counters
  • func NewCounters() *Counters
  • func (s *Counters) All() []Counter
  • func (s *Counters) Decrement(ssid Ssid) (last bool)
  • func (s *Counters) Increment(ssid Ssid, channel []byte) (first bool)
  • func (s *Counters) IncrementOnce(ssid Ssid, channel []byte) (first bool)
• type Frame
  • func DecodeFrame(buf []byte) (out Frame, err error)
  • func NewFrame(capacity int) Frame
  • func (f *Frame) Encode() []byte
  • func (f *Frame) Limit(n int)
  • func (f Frame) Sort()
  • func (f Frame) Split(maxByteSize int) (head Frame, tail Frame)
• type ID
  • func NewID(ssid Ssid) ID
  • func NewPrefix(ssid Ssid, from int64) ID
  • func (id ID) Contract() uint32
  • func (id ID) HasPrefix(ssid Ssid, cutoff int64) bool
  • func (id ID) Match(query Ssid, from, until int64) bool
  • func (id ID) SetTime(t int64)
  • func (id ID) Ssid() Ssid
  • func (id ID) Time() int64
• type Message
  • func DecodeMessage(buf []byte) (out Message, err error)
  • func New(ssid Ssid, channel, payload []byte) *Message
  • func (m *Message) Contract() uint32
  • func (m *Message) Encode() []byte
  • func (m *Message) Expires() time.Time
  • func (m *Message) GetBinaryCodec() binary.Codec
  • func (m *Message) Size() int64
  • func (m *Message) Ssid() Ssid
  • func (m *Message) Stored() bool
  • func (m *Message) Time() int64
• type Ssid
  • func NewSsid(contract uint32, query []uint32) Ssid
  • func NewSsidForPresence(original Ssid) Ssid
  • func NewSsidForShare(original Ssid) Ssid
  • func (s Ssid) Contract() uint32
  • func (s Ssid) Encode() string
  • func (s Ssid) GetHashCode() uint32
• type Subscriber
• type SubscriberType
• type Subscribers
  • func (s *Subscribers) AddRange(from Subscribers, filter func(s Subscriber) bool)
  • func (s *Subscribers) AddUnique(value Subscriber) bool
  • func (s *Subscribers) Contains(value Subscriber) (ok bool)
  • func (s *Subscribers) Random(rnd uint32) (v Subscriber)
  • func (s *Subscribers) Remove(value Subscriber) bool
  • func (s *Subscribers) Reset()
  • func (s *Subscribers) Size() int
• type Subscription
• type Trie
  • func NewTrie() *Trie
  • func NewTrieMQTT() *Trie
  • func (t *Trie) Count() int
  • func (t *Trie) Lookup(ssid Ssid, filter func(s Subscriber) bool) (subs Subscribers)
  • func (t *Trie) Subscribe(ssid Ssid, sub Subscriber) *Subscription
  • func (t *Trie) Unsubscribe(ssid Ssid, subscriber Subscriber)

Constants

const (
	SubscriberDirect = SubscriberType(iota)
	SubscriberRemote
	SubscriberOffline
)

Subscriber types

const RetainedTTL = math.MaxUint32

RetainedTTL represents a TTL value to use for retained messages (max TTL).

Variables

var Query = Ssid{system, query}

Query represents a constant SSID for a query.

Types

type Awaiter

type Awaiter interface {
	Gather(time.Duration) [][]byte
}

Awaiter represents an asynchronously awaiting response channel.

type Counter

type Counter struct {
	Ssid    Ssid
	Channel []byte
	Counter int
}

Counter represents a single subscription counter.

type Counters

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

Counters represents a subscription counting map.

func NewCounters

func NewCounters() *Counters

NewCounters creates a new container.

func (*Counters) All

func (s *Counters) All() []Counter

All returns all counters.

func (*Counters) Decrement

func (s *Counters) Decrement(ssid Ssid) (last bool)

Decrement decrements a subscription counter.

func (*Counters) Increment

func (s *Counters) Increment(ssid Ssid, channel []byte) (first bool)

Increment increments the subscription counter.

func (*Counters) IncrementOnce

func (s *Counters) IncrementOnce(ssid Ssid, channel []byte) (first bool)

IncrementOnce increments the subscription counter.

type Frame

type Frame []Message

Frame represents a message frame which is sent through the wire to the remote server and contains a set of messages.

func DecodeFrame

func DecodeFrame(buf []byte) (out Frame, err error)

DecodeFrame decodes the message frame from the decoder.

func NewFrame

func NewFrame(capacity int) Frame

NewFrame creates a new frame with the specified capacity

func (*Frame) Encode

func (f *Frame) Encode() []byte

Encode encodes the message frame

func (*Frame) Limit

func (f *Frame) Limit(n int)

Limit takes the last N elements, sorted by message time

func (Frame) Sort

func (f Frame) Sort()

Sort sorts the frame

func (Frame) Split

func (f Frame) Split(maxByteSize int) (head Frame, tail Frame)

Split splits the frame by a specified number of bytes into two slices.

type ID

type ID []byte

ID represents a message ID encoded at 128bit and lexigraphically sortable

func NewID

func NewID(ssid Ssid) ID

NewID creates a new message identifier for the current time.

func NewPrefix

func NewPrefix(ssid Ssid, from int64) ID

NewPrefix creates a new message identifier only containing the prefix.

func (ID) Contract

func (id ID) Contract() uint32

Contract retrieves the contract from the message ID.

func (ID) HasPrefix

func (id ID) HasPrefix(ssid Ssid, cutoff int64) bool

HasPrefix matches the prefix with the cutoff time.

func (ID) Match

func (id ID) Match(query Ssid, from, until int64) bool

Match matches the mesage ID with SSID and time bounds.

func (ID) SetTime

func (id ID) SetTime(t int64)

SetTime sets the time on the ID, useful for testing.

func (ID) Ssid

func (id ID) Ssid() Ssid

Ssid retrieves the SSID from the message ID.

func (ID) Time

func (id ID) Time() int64

Time gets the time of the key, adjusted.

type Message

type Message struct {
	ID      ID     `json:"id,omitempty"`   // The ID of the message
	Channel []byte `json:"chan,omitempty"` // The channel of the message
	Payload []byte `json:"data,omitempty"` // The payload of the message
	TTL     uint32 `json:"ttl,omitempty"`  // The time-to-live of the message
}

Message represents a message which has to be forwarded or stored.

func DecodeMessage

func DecodeMessage(buf []byte) (out Message, err error)

DecodeMessage decodes the message from the decoder.

func New

func New(ssid Ssid, channel, payload []byte) *Message

New creates a new message structure from the provided SSID, channel and payload.

func (*Message) Contract

func (m *Message) Contract() uint32

Contract retrieves the contract from the message ID.

func (*Message) Encode

func (m *Message) Encode() []byte

Encode encodes the message into a binary & compressed representation.

func (*Message) Expires

func (m *Message) Expires() time.Time

Expires calculates the expiration time.

func (*Message) GetBinaryCodec

func (m *Message) GetBinaryCodec() binary.Codec

GetBinaryCodec retrieves a custom binary codec.

func (*Message) Size

func (m *Message) Size() int64

Size returns the byte size of the message.

func (*Message) Ssid

func (m *Message) Ssid() Ssid

Ssid retrieves the SSID from the message ID.

func (*Message) Stored

func (m *Message) Stored() bool

Stored returns whether the message is or should be stored.

func (*Message) Time

func (m *Message) Time() int64

Time gets the time of the key, adjusted.

type Ssid

type Ssid []uint32

Ssid represents a subscription ID which contains a contract and a list of hashes for various parts of the channel.

func NewSsid

func NewSsid(contract uint32, query []uint32) Ssid

NewSsid creates a new SSID.

func NewSsidForPresence

func NewSsidForPresence(original Ssid) Ssid

NewSsidForPresence creates a new SSID for presence.

func NewSsidForShare

func NewSsidForShare(original Ssid) Ssid

NewSsidForShare creates a new SSID for shared subscriptions.

func (Ssid) Contract

func (s Ssid) Contract() uint32

Contract gets the contract part from SSID.

func (Ssid) Encode

func (s Ssid) Encode() string

Encode encodes the SSID to a binary format

func (Ssid) GetHashCode

func (s Ssid) GetHashCode() uint32

GetHashCode combines the SSID into a single hash.

type Subscriber

type Subscriber interface {
	ID() string
	Type() SubscriberType
	Send(*Message) error
}

Subscriber is a value associated with a subscription.

type SubscriberType

type SubscriberType uint8

SubscriberType represents a type of subscriber

type Subscribers

type Subscribers map[uint32]Subscriber

Subscribers represents a subscriber set which can contain only unique values.

func (*Subscribers) AddRange

func (s *Subscribers) AddRange(from Subscribers, filter func(s Subscriber) bool)

AddRange adds multiple subscribers from an existing list of subscribers, with filter applied.

func (*Subscribers) AddUnique

func (s *Subscribers) AddUnique(value Subscriber) bool

AddUnique adds a subscriber to the set.

func (*Subscribers) Contains

func (s *Subscribers) Contains(value Subscriber) (ok bool)

Contains checks whether a subscriber is in the set.

func (*Subscribers) Random

func (s *Subscribers) Random(rnd uint32) (v Subscriber)

Random picks a random subscriber from the map. The 'rnd' argument must be a 32-bit randomly generated unsigned integer in range of [0, math.MaxUint32).

func (*Subscribers) Remove

func (s *Subscribers) Remove(value Subscriber) bool

Remove removes a subscriber from the set.

func (*Subscribers) Reset

func (s *Subscribers) Reset()

Reset recycles the list of subscribers.

func (*Subscribers) Size

func (s *Subscribers) Size() int

Size returns the size of the subscriber list.

type Subscription

type Subscription struct {
	Ssid       Ssid       // Gets or sets the SSID (parsed channel) for this subscription.
	Subscriber Subscriber // Gets or sets the subscriber for this subscription.
}

Subscription represents a topic subscription.

type Trie

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

Trie represents an efficient collection of subscriptions with lookup capability.

func NewTrie

func NewTrie() *Trie

NewTrie creates a new subscriptions matcher using standard emitter strategy.

func NewTrieMQTT

func NewTrieMQTT() *Trie

NewTrieMQTT creates a new subscriptions matcher using standard MQTT strategy.

func (*Trie) Count

func (t *Trie) Count() int

Count returns the number of subscriptions.

func (*Trie) Lookup

func (t *Trie) Lookup(ssid Ssid, filter func(s Subscriber) bool) (subs Subscribers)

Lookup returns the Subscribers for the given topic.

func (*Trie) Subscribe

func (t *Trie) Subscribe(ssid Ssid, sub Subscriber) *Subscription

Subscribe adds the Subscriber to the topic and returns a Subscription.

func (*Trie) Unsubscribe

func (t *Trie) Unsubscribe(ssid Ssid, subscriber Subscriber)

Unsubscribe removes the Subscription.