core

Documentation

Index

• Constants
• Variables
• type Cmd
• type CmdCode
• type CmdPropagation
• type Connector
  • func NewConnector() *Connector
  • func (cn *Connector) ConnectTo(l Link) error
  • func (cn *Connector) ExecCmd(cmd *Cmd) error
  • func (cn *Connector) GetMsgCh() chan *Message
  • func (cn *Connector) LinkTo([]Link) error
  • func (cn *Connector) Recv(msg *Message) error
  • func (cn *Connector) RouteTo(map[string]Link) error
  • func (cn *Connector) Send(msg *Message) error
  • func (cn *Connector) String() string
• type Link
• type Message
  • func CpMessage(m *Message) *Message
  • func NewMessage(payload []byte) *Message
  • func NewMessageWithAckCh(ackCh chan MsgStatus, meta map[string]interface{}, payload []byte) *Message
  • func NewMessageWithMeta(meta map[string]interface{}, payload []byte) *Message
  • func (m *Message) AckContinue() error
  • func (m *Message) AckDone() error
  • func (m *Message) AckFailed() error
  • func (m *Message) AckInvalid() error
  • func (m *Message) AckPartialSend() error
  • func (m *Message) AckThrottled() error
  • func (m *Message) AckTimedOut() error
  • func (m *Message) AckUnroutable() error
  • func (m *Message) BumpAttempts() error
  • func (m *Message) GetAckCh() chan MsgStatus
  • func (m *Message) GetAttempts() uint32
  • func (m *Message) GetMeta(key string) (interface{}, bool)
  • func (m *Message) GetMetaAll() map[string]interface{}
  • func (m *Message) GetMetaOrDef(key string, def interface{}) (interface{}, bool)
  • func (m *Message) SetMeta(key string, val interface{})
  • func (m *Message) SetMetaAll(extMeta map[string]interface{})
  • func (m *Message) UnsetMeta(key string) (interface{}, bool)
  • func (m *Message) UnsetMetaAll() map[string]interface{}
• type MsgMeta
• type MsgStatus
• type Params
• type RoutingFunc

Constants

const (
	CmdPpgtBtmUp = iota
	CmdPpgtTopDwn
)

Variables

var (
	ErrMsgPartialSend = fmt.Errorf("Partial message send")
	ErrMsgInvalid     = fmt.Errorf("Invalid message format/payload")
	ErrMsgFailed      = fmt.Errorf("Failed message send")
	ErrMsgTimedOut    = fmt.Errorf("Message send timeout")
	ErrMsgUnroutable  = fmt.Errorf("Unroutable message")
	ErrMsgThrottled   = fmt.Errorf("Message throttled")
)

Types

type Cmd

type Cmd struct {
	Code    CmdCode
	Payload []byte
}

type CmdCode

type CmdCode int8

const (
	CmdCodeStart CmdCode = iota
	CmdCodeStop
)

type CmdPropagation

type CmdPropagation uint8

type Connector

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

func NewConnector

func NewConnector() *Connector

func (*Connector) ConnectTo

func (cn *Connector) ConnectTo(l Link) error

func (*Connector) ExecCmd

func (cn *Connector) ExecCmd(cmd *Cmd) error

func (*Connector) GetMsgCh

func (cn *Connector) GetMsgCh() chan *Message

func (*Connector) LinkTo

func (cn *Connector) LinkTo([]Link) error

func (*Connector) Recv

func (cn *Connector) Recv(msg *Message) error

func (*Connector) RouteTo

func (cn *Connector) RouteTo(map[string]Link) error

func (*Connector) Send

func (cn *Connector) Send(msg *Message) error

func (*Connector) String

func (cn *Connector) String() string

type Link

type Link interface {
	String() string
	Recv(*Message) error
	Send(*Message) error
	ConnectTo(Link) error
	LinkTo([]Link) error
	RouteTo(map[string]Link) error
	ExecCmd(*Cmd) error
}

type Message

type Message struct {
	Payload []byte
	// contains filtered or unexported fields
}

func CpMessage

func CpMessage(m *Message) *Message

func NewMessage

func NewMessage(payload []byte) *Message

func NewMessageWithAckCh

func NewMessageWithAckCh(ackCh chan MsgStatus, meta map[string]interface{}, payload []byte) *Message

func NewMessageWithMeta

func NewMessageWithMeta(meta map[string]interface{}, payload []byte) *Message

func (*Message) AckContinue

func (m *Message) AckContinue() error

func (*Message) AckDone

func (m *Message) AckDone() error

func (*Message) AckFailed

func (m *Message) AckFailed() error

func (*Message) AckInvalid

func (m *Message) AckInvalid() error

func (*Message) AckPartialSend

func (m *Message) AckPartialSend() error

func (*Message) AckThrottled

func (m *Message) AckThrottled() error

func (*Message) AckTimedOut

func (m *Message) AckTimedOut() error

func (*Message) AckUnroutable

func (m *Message) AckUnroutable() error

func (*Message) BumpAttempts

func (m *Message) BumpAttempts() error

func (*Message) GetAckCh

func (m *Message) GetAckCh() chan MsgStatus

func (*Message) GetAttempts

func (m *Message) GetAttempts() uint32

func (*Message) GetMeta

func (m *Message) GetMeta(key string) (interface{}, bool)

func (*Message) GetMetaAll

func (m *Message) GetMetaAll() map[string]interface{}

func (*Message) GetMetaOrDef

func (m *Message) GetMetaOrDef(key string, def interface{}) (interface{}, bool)

func (*Message) SetMeta

func (m *Message) SetMeta(key string, val interface{})

func (*Message) SetMetaAll

func (m *Message) SetMetaAll(extMeta map[string]interface{})

func (*Message) UnsetMeta

func (m *Message) UnsetMeta(key string) (interface{}, bool)

func (*Message) UnsetMetaAll

func (m *Message) UnsetMetaAll() map[string]interface{}

type MsgMeta

type MsgMeta sync.Map

type MsgStatus

type MsgStatus = uint8

const (
	MsgStatusNew MsgStatus = iota
	MsgStatusDone
	MsgStatusPartialSend
	MsgStatusInvalid
	MsgStatusFailed
	MsgStatusTimedOut
	MsgStatusUnroutable
	MsgStatusThrottled
)

type Params

type Params map[string]interface{}

type RoutingFunc

type RoutingFunc func(*Message) (string, error)