core

Documentation

Index

• Constants
• func RegisterChannel(name string, f func(conf interface{}) (IChannel, error))
• func RegisterMultiplexer(name string, f func(conf interface{}) (IMultiplexer, error))
• func RegisterProcessor(name string, f func(conf interface{}) (IProcessor, error))
• func RegisterSelector(name string, f func(conf interface{}) (ISelector, error))
• func RegisterTransport(name string, f func(conf interface{}) (ITransport, error))
• type Coder
• type IChannel
  • func GetChannel(name string, conf interface{}) (IChannel, error)
• type IChannelConsumer
• type IConfig
• type IMsgHandler
• type IMultiplexer
  • func GetMultiplexer(name string, conf interface{}) (IMultiplexer, error)
• type IProcessor
  • func GetProcessor(name, typeName string, conf interface{}) (IProcessor, error)
• type ISelector
  • func GetSelector(name string, conf interface{}) (ISelector, error)
• type ITransport
  • func GetTransport(name, typeName string, conf interface{}) (ITransport, error)
• type InputMsg
• type ProcessorMsg
• type TransportPubMsg
  • func (msg *TransportPubMsg) Encode() ([]byte, error)

Constants

const (
	FORMAT_JSON = int8(0)
	FORMAT_GOB  = int8(1)
	FORMAT_CAPN = int8(2)
)

const CHANNEL_PUB = 0

const CHANNEL_SUB = 1

Functions

func RegisterChannel

func RegisterChannel(name string, f func(conf interface{}) (IChannel, error))

func RegisterMultiplexer

func RegisterMultiplexer(name string, f func(conf interface{}) (IMultiplexer, error))

func RegisterProcessor

func RegisterProcessor(name string, f func(conf interface{}) (IProcessor, error))

func RegisterSelector

func RegisterSelector(name string, f func(conf interface{}) (ISelector, error))

func RegisterTransport

func RegisterTransport(name string, f func(conf interface{}) (ITransport, error))

Types

type Coder

type Coder interface {
	Encode() ([]byte, error)
	Decode(input []byte) error
}

type IChannel

type IChannel interface {
	Init(logProf bool)
	SetTopic(topic string)
	SetGroup(group string)
	SetID(id string)
	GetID() string
	SetDir(dir int)
	GetDir() int
	SetHandler(handler IChannelConsumer)
	Commit(rawMsg []interface{}) error

	Send(topic string, partition int32, keys, bytes []byte, headers map[string]string) error
	SendAndRecv(topic string, partition int32, keys, bytes []byte, headers map[string]string) error
	SendWithRetry(topic string, partition int32, keys, bytes []byte, headers map[string]string) error
	SendAndRecvWithRetry(topic string, partition int32, keys, bytes []byte, headers map[string]string) error
	SendRecv(topic string, bytes []byte, timeout int, headers map[string]string) ([]byte, error)
	PreStart(broker string, statsInterval int)
	Start()
	Stop()
	Close()
}

func GetChannel

func GetChannel(name string, conf interface{}) (IChannel, error)

type IChannelConsumer

type IChannelConsumer interface {
	OnMessage(ctx context.Context, topic string, partition int32, data []byte, rawMsg interface{})
}

type IConfig

type IConfig interface {
	GetDBConfig(name string) (driver string, createAddr string, addr string, persistUnderlying string, persistName string, async bool)
}

type IMsgHandler

type IMsgHandler interface {
}

type IMultiplexer

type IMultiplexer interface {
	AddNode(nodeName string, node interface{}) bool
	GetNode(nodeName string) (interface{}, bool)
	GetChannel(nodeName, serviceID string) (interface{}, bool)
	PreStartChannel(nodeName, serviceID string) bool
	SendNode(nodeName, serviceID string, obj interface{}) error
	SendRecvNode(nodeName, serviceID string, obj interface{}) (interface{}, error)
	Send(nodeName, serviceID string, obj interface{}) error
	SendAndRecv(nodeName, serviceID string, obj interface{}) error
	SendWithRetry(nodeName, serviceID string, obj interface{}) error
	SendAndRecvWithRetry(nodeName, serviceID string, obj interface{}) error
	PreStart()
	Start()
}

func GetMultiplexer

func GetMultiplexer(name string, conf interface{}) (IMultiplexer, error)

type IProcessor

type IProcessor interface {
	OnMultiplexerMessage(serviceID string, msg interface{}) (interface{}, error) //as receiver for multiplexer
	OnMessage(topic string, partition int32, data []byte) ([]byte, error)        //as receiver for transport
	PreStart()
	Start()
}

func GetProcessor

func GetProcessor(name, typeName string, conf interface{}) (IProcessor, error)

type ISelector

type ISelector interface {
	GetName() string
	AddNode(node string)
	GetNode(node string) (string, error)
	DelNode(node string) bool
}

func GetSelector

func GetSelector(name string, conf interface{}) (ISelector, error)

type ITransport

type ITransport interface {
	AddChannel(dir int, id, topic, grp string, conf interface{}) bool
	GetChannel(id string) (IChannel, bool)
	PreStartChannel(id string) bool
	StartChannel(id string)
	StopChannel(id string)
	SetID(id string)
	Init(logProf bool)
	SetBrokers(broker string)
	GetBrokers() string
	SetStatsInterval(interval int)
	GetStatsInterval() int
	PreStart()
	Start()
}

func GetTransport

func GetTransport(name, typeName string, conf interface{}) (ITransport, error)

type InputMsg

type InputMsg interface {
	InitFromBytes([]byte)
	GetMsgType() int32
}

type ProcessorMsg

type ProcessorMsg struct {
	MsgType  int32
	Msg      interface{}
	UserID   string
	DeviceID string
}

type TransportPubMsg

type TransportPubMsg struct {
	Partition int32
	Format    int8
	Topic     string
	Keys      []byte
	Obj       interface{}
	Timeout   int
	Inst      int
	Headers   map[string]string
}

func (*TransportPubMsg) Encode

func (msg *TransportPubMsg) Encode() ([]byte, error)