gate

Documentation

Index

• func NewGateHandler(gate Gate) *handler
• type Agent
• type AgentLearner
• type Gate
  • func (this *Gate) GetAgentLearner() AgentLearner
  • func (this *Gate) GetSessionLearner() SessionLearner
  • func (this *Gate) GetStorageHandler() (storage StorageHandler)
  • func (this *Gate) NewSession(data []byte) (Session, error)
  • func (gate *Gate) OnDestroy()
  • func (gate *Gate) OnInit()
  • func (gate *Gate) Run(closeSig chan bool)
  • func (this *Gate) SetSessionLearner(sessionLearner SessionLearner) error
  • func (this *Gate) SetStorageHandler(storage StorageHandler) error
• type GateHandler
• type Session
  • func NewSession(chanRpc *chanrpc.Server, data []byte) (Session, error)
  • func NewSessionByMap(chanRpc *chanrpc.Server, data map[string]interface{}) (Session, error)
• type SessionImp
  • func (*SessionImp) Descriptor() ([]byte, []int)
  • func (m *SessionImp) GetIP() string
  • func (m *SessionImp) GetNetwork() string
  • func (m *SessionImp) GetServerid() string
  • func (m *SessionImp) GetSessionid() string
  • func (m *SessionImp) GetSettings() map[string]string
  • func (m *SessionImp) GetTracer() map[string]string
  • func (m *SessionImp) GetUserid() string
  • func (*SessionImp) ProtoMessage()
  • func (m *SessionImp) Reset()
  • func (m *SessionImp) String() string
  • func (m *SessionImp) XXX_DiscardUnknown()
  • func (m *SessionImp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
  • func (m *SessionImp) XXX_Merge(src proto.Message)
  • func (m *SessionImp) XXX_Size() int
  • func (m *SessionImp) XXX_Unmarshal(b []byte) error
• type SessionLearner
• type StorageHandler

Functions

func NewGateHandler

func NewGateHandler(gate Gate) *handler

Types

type Agent

type Agent interface {
	WriteMsg(msg interface{})
	WriteMsgByte(data [][]byte)
	LocalAddr() net.Addr
	RemoteAddr() net.Addr
	Close()
	Destroy()
	UserData() interface{}
	SetUserData(data interface{})
	Prob() interface{}
	SetProb(prob interface{})
	GetSession() Session
	SetAgentID(id int64)
	GetAgentID() int64
	IsClosed() bool
	SetTableID(id int)
	GetTableID() int
}

type AgentLearner

type AgentLearner interface {
	Connect(a Agent)    //当连接建立  并且MQTT协议握手成功
	DisConnect(a Agent) //当连接关闭	或者客户端主动发送MQTT DisConnect命令
}

type Gate

type Gate struct {
	MaxConnNum      int
	PendingWriteNum int
	MaxMsgLen       uint32
	Processor       network.Processor
	AgentChanRPC    *chanrpc.Server

	// websocket
	WSAddr      string
	HTTPTimeout time.Duration
	CertFile    string
	KeyFile     string

	// tcp
	TCPAddr      string
	LenMsgLen    int
	LittleEndian bool
	// contains filtered or unexported fields
}

func (*Gate) GetAgentLearner

func (this *Gate) GetAgentLearner() AgentLearner

func (*Gate) GetSessionLearner

func (this *Gate) GetSessionLearner() SessionLearner

func (*Gate) GetStorageHandler

func (this *Gate) GetStorageHandler() (storage StorageHandler)

func (*Gate) NewSession

func (this *Gate) NewSession(data []byte) (Session, error)

func (*Gate) OnDestroy

func (gate *Gate) OnDestroy()

func (*Gate) OnInit

func (gate *Gate) OnInit()

func (*Gate) Run

func (gate *Gate) Run(closeSig chan bool)

func (*Gate) SetSessionLearner

func (this *Gate) SetSessionLearner(sessionLearner SessionLearner) error

* 设置客户端连接和断开的监听器

func (*Gate) SetStorageHandler

func (this *Gate) SetStorageHandler(storage StorageHandler) error

* 设置Session信息持久化接口

type GateHandler

type GateHandler interface {
	Bind(args []interface{}) interface{}      //Bind the session with the the Userid.
	UnBind(args []interface{}) interface{}    //UnBind the session with the the Userid.
	Set(args []interface{}) interface{}       //Set values (one or many) for the session.
	Remove(args []interface{}) interface{}    //Remove value from the session.
	Push(args []interface{}) interface{}      //推送信息给Session
	Send(args []interface{}) interface{}      //Send message
	SendBatch(args []interface{}) interface{} //批量发送
	BroadCast(args []interface{}) interface{} //广播消息给网关所有在连客户端
	//查询某一个userId是否连接中，这里只是查询这一个网关里面是否有userId客户端连接，如果有多个网关就需要遍历了
	IsConnect(args []interface{}) interface{}
	Close(args []interface{}) interface{}             //主动关闭连接
	Update(args []interface{}) interface{}            //更新整个Session 通常是其他模块拉取最新数据
	OnDestory()                                       //退出事件,主动关闭所有的连接
	CloseMultiSession(args []interface{}) interface{} //關閉重複登入的連線
}

* net代理服务 处理器

type Session

type Session interface {
	GetIP() string
	GetNetwork() string
	GetUserid() string
	GetSessionid() string
	GetServerid() string
	GetSettings() map[string]string
	SetIP(ip string)
	SetNetwork(network string)
	SetUserid(userid string)
	SetSessionid(sessionid string)
	SetServerid(serverid string)
	SetSettings(settings map[string]string)
	Serializable() ([]byte, error)
	Update() (err error)
	Bind(Userid string) string
	UnBind() string
	Push() string
	Set(key string, value string) string
	SetPush(key string, value string) string //设置值以后立即推送到gate网关
	Get(key string) string
	Remove(key string) string
	Send(id string, data interface{}) string
	SendNR(id string, data interface{}) string
	SendBatch(Sessionids string, data interface{}) string //想该客户端的网关批量发送消息
	//查询某一个userId是否连接中，这里只是查询这一个网关里面是否有userId客户端连接，如果有多个网关就需要遍历了
	IsConnect(Userid string) (bool, string)
	//是否是访客(未登录) ,默认判断规则为 userId==""代表访客
	IsGuest() bool
	//设置自动的访客判断函数,记得一定要在全局的时候设置这个值,以免部分模块因为未设置这个判断函数造成错误的判断
	JudgeGuest(judgeGuest func(session Session) bool)
	Close() (err error)
	CloseMultiSession(key string) string
}

func NewSession

func NewSession(chanRpc *chanrpc.Server, data []byte) (Session, error)

func NewSessionByMap

func NewSessionByMap(chanRpc *chanrpc.Server, data map[string]interface{}) (Session, error)

type SessionImp

type SessionImp struct {
	IP                   string            `protobuf:"bytes,1,opt,name=IP,proto3" json:"IP,omitempty"`
	Network              string            `protobuf:"bytes,2,opt,name=Network,proto3" json:"Network,omitempty"`
	Userid               string            `protobuf:"bytes,3,opt,name=Userid,proto3" json:"Userid,omitempty"`
	Sessionid            string            `protobuf:"bytes,4,opt,name=Sessionid,proto3" json:"Sessionid,omitempty"`
	Serverid             string            `protobuf:"bytes,5,opt,name=Serverid,proto3" json:"Serverid,omitempty"`
	Settings             map[string]string `` /* 157-byte string literal not displayed */
	Tracer               map[string]string `` /* 153-byte string literal not displayed */
	XXX_NoUnkeyedLiteral struct{}          `json:"-"`
	XXX_unrecognized     []byte            `json:"-"`
	XXX_sizecache        int32             `json:"-"`
}

func (*SessionImp) Descriptor

func (*SessionImp) Descriptor() ([]byte, []int)

func (*SessionImp) GetIP

func (m *SessionImp) GetIP() string

func (*SessionImp) GetNetwork

func (m *SessionImp) GetNetwork() string

func (*SessionImp) GetServerid

func (m *SessionImp) GetServerid() string

func (*SessionImp) GetSessionid

func (m *SessionImp) GetSessionid() string

func (*SessionImp) GetSettings

func (m *SessionImp) GetSettings() map[string]string

func (*SessionImp) GetTracer

func (m *SessionImp) GetTracer() map[string]string

func (*SessionImp) GetUserid

func (m *SessionImp) GetUserid() string

func (*SessionImp) ProtoMessage

func (*SessionImp) ProtoMessage()

func (*SessionImp) Reset

func (m *SessionImp) Reset()

func (*SessionImp) String

func (m *SessionImp) String() string

func (*SessionImp) XXX_DiscardUnknown

func (m *SessionImp) XXX_DiscardUnknown()

func (*SessionImp) XXX_Marshal

func (m *SessionImp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*SessionImp) XXX_Merge

func (m *SessionImp) XXX_Merge(src proto.Message)

func (*SessionImp) XXX_Size

func (m *SessionImp) XXX_Size() int

func (*SessionImp) XXX_Unmarshal

func (m *SessionImp) XXX_Unmarshal(b []byte) error

type SessionLearner

type SessionLearner interface {
	Connect(a Agent)    //当连接建立  并且MQTT协议握手成功
	DisConnect(a Agent) //当连接关闭	或者客户端主动发送MQTT DisConnect命令
}

type StorageHandler

type StorageHandler interface {
	/**
	存储用户的Session信息
	Session Bind Userid以后每次设置 settings都会调用一次Storage
	*/
	Storage(Userid string, session Session) (err error)
	/**
	强制删除Session信息
	*/
	Delete(Userid string) (err error)
	/**
	获取用户Session信息
	Bind Userid时会调用Query获取最新信息
	*/
	Query(Userid string) (data []byte, err error)
	/**
	用户心跳,一般用户在线时1s发送一次
	可以用来延长Session信息过期时间
	*/
	Heartbeat(Userid string)
	//剔除重複登入者
	CloseMultiSession(Userid string)
}

* Session信息持久化