pigeon

type AcceptHandleFunc ¶

type AcceptHandleFunc func(conn net.Conn)

type Acceptor ¶

type Acceptor interface {
	Lifecycle
	Sync
}

Acceptor is a interface wraps necessary methods for network connection acceptance. The implementation should be based on FSM.

func NewAcceptor(config AcceptorConfig) Acceptor

type AcceptorConfig ¶

type AcceptorConfig struct {
	Parallelism      uint8
	Listener         net.Listener
	AcceptHandleFunc func(conn net.Conn)
}

AcceptorConfig is a data struct for acceptor initialization.

type AckManager ¶

type AckManager interface {
	// InitAck init and register a ack transaction to manager.
	InitAck(key interface{})

	// WaitAck will block invoker goroutine until specified ack transaction commit or timeout.
	WaitAck(key interface{}, timeout time.Duration) (data interface{}, err error)

	// CommitAck commit specified ack transaction.
	CommitAck(key interface{}, data interface{})
}

AckManager is the interface wraps methods for acknowledgement management.

func NewAckManager() AckManager

type ApolloConfig ¶

type ApolloConfig struct {
	TLVConfig
	// contains filtered or unexported fields
}

func (c *ApolloConfig) RegisterEntity(constructor func() ApolloEntity)

type ApolloEntity ¶

type ApolloEntity interface {
	TypeCode() uint16
}

type ByteBuf ¶

type ByteBuf interface {
	io.Writer
	io.Reader

	ReadIndex() int
	ReadBytes(length int) []byte
	ReadableBytes() int

	WriteIndex() int
	WriteBytes(bytes []byte)
	WritableBytes() int

	Capacity() int
	Reset()
	Release()
}

ByteBuf is the interface provide necessary method of byte buffer with double indexes. The implementation must be designed to consist of two indexes that are read and written separately.

The relationship between reading index and writing index:

+--------------+--------------------+------------------+
|  read bytes  |   readable bytes   |  writable bytes  |
|              |  (unread content)  |      (free)      |
+--------------+--------------------+------------------+
↑              ↑                    ↑                  ↑
0    <=   readerIndex    <=    writerIndex    <=    capacity
+------------------------------------------------------+

Method Reset will reset both read index and write index to 0. Method Release will release read bytes and recalculate indexes.

func NewByteBuf(initSize int) ByteBuf

type Channel ¶

type Channel interface {
	Sender
	Close
	Remote() net.Addr
	IsConnected() bool
	GetContext(key string) interface{}
	AddContext(key string, val interface{})
	DelContext(key string)
}

func NewChannel(pipeline Pipeline) Channel

type ChannelGroup ¶

type ChannelGroup interface {
	// Add will add a specified channel to channel group.
	Add(channel Channel)

	// Remove will remove specified channel from channel group.
	Remove(channel Channel)

	// CloseAll will close all channel which management by channel group and remove
	// all channel from channel group.
	CloseAll()
}

ChannelGroup is a interface wraps methods for channel management which provide batch close support for channels.

func NewHashSafeChannelGroup() ChannelGroup

type ChannelHandler ¶

type ChannelHandler interface {
	// ChannelActivate will be invoked while connection is ready.
	ChannelActivate(channel Channel) error

	// ChannelInActivate will be invoked after connection closed.
	ChannelInactivate(channel Channel) error

	// ChannelRead will be invoked while a message is ready.
	ChannelRead(channel Channel, in interface{}) error

	// ChannelError will be invoked while some exception happened.
	ChannelError(channel Channel, channelErr error)
}

ChannelHandler is the interface provide necessary methods for channel event handle which invoked by pipeline.

type Client ¶

type Client interface {
	Lifecycle
	Sync
	Sender
}

Client is the interface that wraps the basic method to implement a tcp network client.

func NewClient(cfg ClientConfig, initializer PipelineInitializer) Client

type ClientConfig ¶

type ClientConfig struct {
	TCPConfig
	Timeout time.Duration
}

ClientConfig provide properties for client configuration

type Close ¶

type Close interface {
	Close()
}

Close is the interface which wraps method Close().

type DecodeError ¶

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

DecodeError is a error implementation with detail error string output include decoder name and cause for decode exception. The format of complete error string is '$DECODER decode error cause $CAUSE'.

func (e *DecodeError) Error() string

type DecoderInitFunc ¶

type DecoderInitFunc func() FrameDecoder

type EncodeError ¶

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

EncodeError is a error implementation with detail error string output include encoder name and cause for encode exception. The format of complete error string is '$ENCODER encode error cause $CAUSE'.

func (e *EncodeError) Error() string

type EncoderInitFunc ¶

type EncoderInitFunc func() FrameEncoder

type FrameCodec ¶

type FrameCodec interface {
	FrameDecoder
	FrameEncoder
}

FrameCodec is the interface that wraps the basic method for both encode and decode.

type FrameDecoder ¶

type FrameDecoder interface {
	Decode(in ByteBuf) (result interface{}, err error)
}

FrameDecoder is the interface that wraps the basic method for decode tcp stream. A FrameDecoder will be instantiated and init by PipelineInitializer in Pipeline initializing.

Model:

+-------------------------+
|     (src)↓              |
|  Decode(in) → (result)  |
+-------------------------+

func NewApolloFrameDecoder(config ApolloConfig) FrameDecoder

func NewStringFrameDecoder() FrameDecoder

func NewTLVFrameDecoder(config TLVConfig) FrameDecoder

type FrameEncoder ¶

type FrameEncoder interface {
	Encode(msg interface{}) (result []byte, err error)
}

FrameDecoder is the interface that wraps the basic method for encode tcp stream. A FrameEncoder will be instantiated and init by PipelineInitializer in Pipeline initializing.

Model:

+-------------------------+
|     (src)↓              |
|  Encode(in) → (result)  |
+-------------------------+

func NewApolloFrameEncoder(config ApolloConfig) FrameEncoder

func NewStringFrameEncoder() FrameEncoder

func NewTLVFrameEncoder(config TLVConfig) FrameEncoder

type FunctionalChannelHandler ¶

type FunctionalChannelHandler struct {
	HandleActivate   func(channel Channel) error
	HandleInactivate func(channel Channel) error
	HandleRead       func(channel Channel, in interface{}) error
	HandleError      func(channel Channel, err error)
}

FunctionalChannelHandler is a public implementation of ChannelHandler interface which support functional definition for business logic.

func (h *FunctionalChannelHandler) ChannelActivate(channel Channel) error

func (h *FunctionalChannelHandler) ChannelError(channel Channel, channelErr error)

func (h *FunctionalChannelHandler) ChannelInactivate(channel Channel) error

func (h *FunctionalChannelHandler) ChannelRead(channel Channel, in interface{}) error

type FunctionalPipelineInitializer ¶

type FunctionalPipelineInitializer struct {
	DecoderInit DecoderInitFunc
	EncoderInit EncoderInitFunc
	HandlerInit HandlerInitFunc
}

FunctionalPipelineInitializer is a public implementation of PipelineInitializer interface which support functional definition for pipeline initialization logic.

func (i *FunctionalPipelineInitializer) InitDecoder() FrameDecoder

func (i *FunctionalPipelineInitializer) InitEncoder() FrameEncoder

func (i *FunctionalPipelineInitializer) InitHandler() ChannelHandler

type HandlerInitFunc ¶

type HandlerInitFunc func() ChannelHandler

type Init ¶

type Init interface {
	Init() error
}

Init is the interface that wraps the basic method for init.

type Level ¶

type Level uint8

type Lifecycle ¶

type Lifecycle interface {
	Start() error
	Stop()
	IsRunning() bool
}

LifeCycle is the interface that wraps the basic method for standard service lifecycle control.

type Logger ¶

type Logger interface {
	Trace(format string, args ...interface{})
	Debug(format string, args ...interface{})
	Info(format string, args ...interface{})
	Warn(format string, args ...interface{})
	Error(format string, args ...interface{})
}

type Pipeline ¶

type Pipeline interface {
	Lifecycle
	Sync
	Sender
	GetChannel() Channel
	Remote() net.Addr
}

Pipeline is the interface defined necessary methods which makes a pipeline of FrameDecoder, FrameEncoder, and ChannelHandler for inbound and outbound data processing.

Model:

+---------------------------------------+
|         TCP Network Connection        |
+---------------------------------------+
        ↑(write)               ↓(read)
+----------------+     +----------------+
|  FrameEncoder  |     |  FrameDecoder  |
+----------------+     +----------------+
        ↑(outbound)            ↓(inbound)
+----------------+     +----------------+
|    Channel     |     | ChannelHandler |
+----------------+     +----------------+

State:

+-----+           +---------+          +----------+
| NEW | → Start → | RUNNING | → Stop → | SHUTDOWN |
+-----+           +---------+          +----------+

Notes: The implementations should be parallel safe and based on FSM.

func NewPipeline(conn net.Conn, initializer PipelineInitializer) (Pipeline, error)

type PipelineInitializer ¶

type PipelineInitializer interface {
	//  InitDecoder used for decoder initialization.
	InitDecoder() FrameDecoder

	//  InitEncoder used for encoder initialization.
	InitEncoder() FrameEncoder

	// InitHandler used for channel handler initialization.
	InitHandler() ChannelHandler
}

ChannelHandler is the interface provide necessary methods for pipeline initialization which invoked by pipeline.

type Sender ¶

type Sender interface {
	Send(data interface{}) error
	SendFuture(data interface{}, callback func(err error))
}

type Server ¶

type Server interface {
	Lifecycle
	Sync
}

Server is the interface that wraps the basic method to implement a tcp network server based on FSM.

func NewServer(cfg ServerConfig, initializer PipelineInitializer) Server

type ServerConfig ¶

type ServerConfig struct {
	TCPConfig
	AcceptorSize uint8
}

ServerConfig provide properties for server configuration

type Sync ¶

type Sync interface {
	Sync()
}

Sync is the interface that wraps the Sync() method which will block invoker goroutine.

type TCPConfig ¶

type TCPConfig struct {
	Port            int
	IP              net.IP
	KeepAlive       bool
	KeepAlivePeriod time.Duration
}

func (c *TCPConfig) ApplyTCP(conn *net.TCPConn)

type TLVConfig ¶

type TLVConfig struct {
	TagValue   uint8
	FrameLimit uint32
}

TLVConfig is a data struct provide configuration properties for both tlvFrameDecoder and tlvFrameEncoder.

+----------+-----------+-----------+
|    TAG   |  LENGTH   |   VALUE   |
| (1 byte) | (4 bytes) | (payload) |
+----------+-----------+-----------+
     ↑
  TagValue

type Type ¶

type Type interface {
	Type() string
}

Type is the interface wraps method to get string type value.

type UnknownAddr ¶

type UnknownAddr struct {
}

func (ua *UnknownAddr) Network() string

func (ua *UnknownAddr) String() string