him

Documentation

Index

• Constants
• Variables
• type Acceptor
• type Agent
• type Channel
  • func NewChannel(id string, conn Conn) Channel
• type ChannelMap
  • func NewChannelMap(num int) ChannelMap
• type ChannelMapImpl
  • func (c *ChannelMapImpl) Add(channel Channel)
  • func (c *ChannelMapImpl) All() []Channel
  • func (c *ChannelMapImpl) Get(id string) (Channel, bool)
  • func (c *ChannelMapImpl) Remove(id string)
• type Client
• type Conn
• type Context
  • func BuildContext() Context
• type ContextImpl
  • func (c *ContextImpl) Dispatch(body proto.Message, recs ...*Location) error
  • func (c *ContextImpl) Header() *pkt.Header
  • func (c *ContextImpl) Next()
  • func (c *ContextImpl) ReadBody(val proto.Message) error
  • func (c *ContextImpl) Resp(status pkt.Status, body proto.Message) error
  • func (c *ContextImpl) RespWithError(status pkt.Status, err error) error
  • func (c *ContextImpl) Session() Session
• type Dialer
• type DialerContext
• type Dispatcher
• type Event
  • func NewEvent() *Event
  • func (e *Event) Done() <-chan struct{}
  • func (e *Event) Fire() bool
  • func (e *Event) HasFired() bool
• type Frame
• type FuncTree
  • func NewTree() *FuncTree
  • func (t *FuncTree) Add(path string, handlers ...HandlerFunc)
  • func (t *FuncTree) Get(path string) (HandlersChain, bool)
• type Handler
• type HandlerFunc
• type HandlersChain
• type Location
  • func (loc *Location) Bytes() []byte
  • func (loc *Location) Unmarshal(data []byte) (err error)
• type MessageListener
• type OpCode
• type Router
  • func NewRouter() *Router
  • func (r *Router) AddHandles(command string, handlers ...HandlerFunc)
  • func (r *Router) Serve(pkt *pkt.LogicPkt, dispatcher Dispatcher, cache SessionStorage, ...) error
• type Server
• type Service
• type ServiceRegistration
• type Session
• type SessionStorage
• type StateListener

Constants

const (
	DefaultReadWait  = time.Minute * 3
	DefaultWriteWait = time.Second * 10
	DefaultLoginWait = time.Second * 10
	DefaultHeartbeat = time.Second * 55
)

Variables

var ErrSessionNil = errors.New("err:session nil")

ErrNil

Types

type Acceptor

type Acceptor interface {
	// Accept 返回一个握手完成的Channel对象或者一个error。
	// 业务层需要处理不同协议和网络环境的下连接握手协议
	Accept(Conn, time.Duration) (string, error)
}

Acceptor 连接接收器

type Agent

type Agent interface {
	// ID 返回连接的channelID
	ID() string
	// Push 用于上层业务返回消息
	Push([]byte) error
}

Agent 表示发送方

type Channel

type Channel interface {
	Conn
	Agent
	Close() error //overwrite net.Conn.Close()
	ReadLoop(MessageListener) error
	SetWriteWait(time.Duration)
	SetReadWait(time.Duration)
}

Channel 上层通用逻辑的封装

func NewChannel

func NewChannel(id string, conn Conn) Channel

type ChannelMap

type ChannelMap interface {
	Add(Channel)
	Remove(id string)
	Get(id string) (Channel, bool)
	All() []Channel
}

ChannelMap ChannelMap

func NewChannelMap

func NewChannelMap(num int) ChannelMap

type ChannelMapImpl

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

func (*ChannelMapImpl) Add

func (c *ChannelMapImpl) Add(channel Channel)

Add channel to channelMap

func (*ChannelMapImpl) All

func (c *ChannelMapImpl) All() []Channel

func (*ChannelMapImpl) Get

func (c *ChannelMapImpl) Get(id string) (Channel, bool)

func (*ChannelMapImpl) Remove

func (c *ChannelMapImpl) Remove(id string)

type Client

type Client interface {
	Service
	Connect(string) error
	SetDialer(Dialer)
	Send([]byte) error
	Read() (Frame, error)
	Close()
}

type Conn

type Conn interface {
	net.Conn
	ReadFrame() (Frame, error)
	WriteFrame(OpCode, []byte) error
	Flush() error
}

Conn 对net.Conn进行二次封装，把读与写的操作封装到连接中

type Context

type Context interface {
	Dispatcher
	SessionStorage
	Header() *pkt.Header
	ReadBody(val proto.Message) error
	Session() Session
	RespWithError(status pkt.Status, err error) error
	Resp(status pkt.Status, body proto.Message) error
	Dispatch(body proto.Message, revs ...*Location) error
	Next()
}

Context 实现处理器中的上下文

func BuildContext

func BuildContext() Context

type ContextImpl

type ContextImpl struct {
	sync.Mutex
	Dispatcher
	SessionStorage
	// contains filtered or unexported fields
}

ContextImpl is the most important part of him

func (*ContextImpl) Dispatch

func (c *ContextImpl) Dispatch(body proto.Message, recs ...*Location) error

Dispatch 派发消息到指定的接收方

func (*ContextImpl) Header

func (c *ContextImpl) Header() *pkt.Header

func (*ContextImpl) Next

func (c *ContextImpl) Next()

Next 执行责任链

func (*ContextImpl) ReadBody

func (c *ContextImpl) ReadBody(val proto.Message) error

func (*ContextImpl) Resp

func (c *ContextImpl) Resp(status pkt.Status, body proto.Message) error

Resp 回复消息，也就是给发送方返回一条消息

func (*ContextImpl) RespWithError

func (c *ContextImpl) RespWithError(status pkt.Status, err error) error

func (*ContextImpl) Session

func (c *ContextImpl) Session() Session

type Dialer

type Dialer interface {
	DialAndHandshake(DialerContext) (net.Conn, error)
}

type DialerContext

type DialerContext struct {
	Id      string
	Name    string
	Address string
	Timeout time.Duration
}

type Dispatcher

type Dispatcher interface {
	Push(gateway string, channels []string, p *pkt.LogicPkt) error
}

Dispatcher 向网关中的channels两个连接推送一条消息LogicPkt 这个能力由容器提供

type Event

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

func NewEvent

func NewEvent() *Event

func (*Event) Done

func (e *Event) Done() <-chan struct{}

func (*Event) Fire

func (e *Event) Fire() bool

Fire causes e to complete. It is safe to call multiple times, and concurrently. It returns true iff this call to Fire caused the signaling channel returned by Done to close. 用于触发事件，它通过sync.Once确保事件只会被触发一次， 使用atomic.StoreInt32()原子操作将fired标志位置为1，表示事件已经触发。 同时，它还会关闭通道c，以通知等待该事件的goroutine

func (*Event) HasFired

func (e *Event) HasFired() bool

type Frame

type Frame interface {
	SetOpCode(OpCode)
	GetOpCode() OpCode
	SetPayload([]byte)
	GetPayload() []byte
}

Frame 通过抽象一个Frame接口解决底层封包与拆包

type FuncTree

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

func NewTree

func NewTree() *FuncTree

NewTree New Tree

func (*FuncTree) Add

func (t *FuncTree) Add(path string, handlers ...HandlerFunc)

func (*FuncTree) Get

func (t *FuncTree) Get(path string) (HandlersChain, bool)

type Handler

type Handler interface {
	MessageListener
	Acceptor
	StateListener
}

type HandlerFunc

type HandlerFunc func(ctx Context)

HandlerFunc defines the handler used 处理器

type HandlersChain

type HandlersChain []HandlerFunc

HandlersChain 处理链

type Location

type Location struct {
	// 通道Id
	ChannelId string
	// 网关Id
	GateId string
}

Location 用户位置，

func (*Location) Bytes

func (loc *Location) Bytes() []byte

func (*Location) Unmarshal

func (loc *Location) Unmarshal(data []byte) (err error)

type MessageListener

type MessageListener interface {
	// Receive 参数Agent表示发送方
	Receive(Agent, []byte)
}

MessageListener 消息监听器

type OpCode

type OpCode byte

OpCode 定义统一的OpCode

const (
	OpContinuation OpCode = 0x0
	OpText         OpCode = 0x1
	OpBinary       OpCode = 0x2
	OpClose        OpCode = 0x8
	OpPing         OpCode = 0x9
	OpPong         OpCode = 0xa
)

type Router

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

func NewRouter

func NewRouter() *Router

func (*Router) AddHandles

func (r *Router) AddHandles(command string, handlers ...HandlerFunc)

AddHandles 添加handlers

func (*Router) Serve

func (r *Router) Serve(pkt *pkt.LogicPkt, dispatcher Dispatcher, cache SessionStorage, session Session) error

type Server

type Server interface {
	ServiceRegistration
	SetAcceptor(Acceptor)
	SetMessageListener(MessageListener)
	SetStateListener(StateListener)
	SetReadWait(time.Duration)
	SetChannelMap(ChannelMap)

	Start() error
	Push(string, []byte) error
	Shutdown(ctx context.Context) error
}

Server 服务器用于承载Service

type Service

type Service interface {
	ServiceID() string
	ServiceName() string
	GetMeta() map[string]string
}

Service 定义基础服务的抽象接口

type ServiceRegistration

type ServiceRegistration interface {
	Service
	// PublicAddress ip or domain
	PublicAddress() string
	PublicPort() int
	DialURL() string
	GetProtocol() string
	GetNamespace() string
	GetTags() []string
	// String SetTags(tags []string)
	// SetMeta(meta map[string]string)
	String() string
}

ServiceRegistration Service define a Service

type Session

type Session interface {
	GetChannelId() string
	GetGateId() string
	GetAccount() string
	GetZone() string
	GetIsp() string
	GetRemoteIP() string
	GetDevice() string
	GetApp() string
	GetTags() []string
}

Session 会话

type SessionStorage

type SessionStorage interface {
	// Add a session
	Add(session *pkt.Session) error
	// Delete a session
	Delete(account string, channelId string) error
	// Get session by channelId
	Get(channelId string) (*pkt.Session, error)
	// GetLocations Get Locations by accounts
	GetLocations(account ...string) ([]*Location, error)
	// GetLocation Get Location by account and device
	GetLocation(account string, device string) (*Location, error)
}

SessionStorage 定义会话存储，提供基于保存、删除、查找会话的功能

type StateListener

type StateListener interface {
	// Disconnect 连接断开回调
	Disconnect(string) error
}

StateListener 状态监听器