pubsub

Documentation

Index

• Constants
• Variables
• func NewEndpoint(config *EndpointConfig, exchange pubsub.Exchange) (pubsub.Endpoint, error)
• type Endpoint
  • func (ep *Endpoint) Close() error
  • func (ep *Endpoint) Publish(event pubsub.Event) error
• type EndpointConfig
  • func NewEndpointConfig() *EndpointConfig
  • func (config *EndpointConfig) FeedFromEnv(prefix string) error
  • func (config *EndpointConfig) IsComplete() bool
  • func (config *EndpointConfig) MustBeComplete()
  • func (config *EndpointConfig) MustFeedFromEnv(prefix string) *EndpointConfig
• type Event
  • func (e Event) Body() []byte
  • func (e Event) Kind() []byte
  • func (e Event) Publisher() []byte
  • func (e Event) Seq() []byte
  • func (e Event) SetSeq(seq pubsub.EventSeqNum)

Constants

const (
	MessageTypeEvent byte = iota
	MessageTypeSeqTable
)

const Header = "CDR#PUBSUB@01"

Variables

var (
	FrameHeader = []byte(Header)

	FrameEventType    = []byte{MessageTypeEvent}
	FrameSeqTableType = []byte{MessageTypeSeqTable}
)

var (
	ErrConfigIncomplete = errors.New("PubSub endpoint is not fully configured")
	ErrSeqNumNotSet     = errors.New("Event sequence number is not set")
)

Functions

func NewEndpoint

func NewEndpoint(config *EndpointConfig, exchange pubsub.Exchange) (pubsub.Endpoint, error)

Types

type Endpoint

type Endpoint struct {
	*loop.MessageLoop
	// contains filtered or unexported fields
}

func (*Endpoint) Close

func (ep *Endpoint) Close() error

func (*Endpoint) Publish

func (ep *Endpoint) Publish(event pubsub.Event) error

type EndpointConfig

type EndpointConfig struct {
	RouterEndpoint string
	RouterSndhwm   int
	RouterRcvhwm   int
	PubSndhwm      int
	PubEndpoint    string
}

func NewEndpointConfig

func NewEndpointConfig() *EndpointConfig

func (*EndpointConfig) FeedFromEnv

func (config *EndpointConfig) FeedFromEnv(prefix string) error

func (*EndpointConfig) IsComplete

func (config *EndpointConfig) IsComplete() bool

func (*EndpointConfig) MustBeComplete

func (config *EndpointConfig) MustBeComplete()

func (*EndpointConfig) MustFeedFromEnv

func (config *EndpointConfig) MustFeedFromEnv(prefix string) *EndpointConfig

type Event

type Event [][]byte

FRAME 0: publisher (string) FRAME 1: event kind (string) FRAME 2: message header (string) FRAME 3: message type (byte) FRAME 4: event sequence number (uint32, BE) FRAME 5: event object, marshalled (bytes)

func (Event) Body

func (e Event) Body() []byte

func (Event) Kind

func (e Event) Kind() []byte

func (Event) Publisher

func (e Event) Publisher() []byte

func (Event) Seq

func (e Event) Seq() []byte

func (Event) SetSeq

func (e Event) SetSeq(seq pubsub.EventSeqNum)