colong

Documentation

Index

• Constants
• Variables
• func AddSession(name string, session getty.Session)
• func ClientRequest()
• func EncodeCMsg(msg WrapCMsg) ([]byte, error)
• func InitClusterTaskPool(poolSize int) (close io.Closer)
• func NewEventListener() getty.EventListener
• func NewServerEventListener() getty.EventListener
• func RemoveSession(name string)
• func SendMsgToCluster(msg messagev5.Message, shareName, targetNode string, ...)
• func SetCurName(listener getty.EventListener, curName string)
• func SetLoggerLevelInfo()
• func SetPubFunc(el getty.EventListener, clusterInToPub ClusterInToPub, ...)
• func SetSessionOnOpen(listener getty.EventListener, fuc func(name string, session getty.Session))
• func UpdateLogger(lg getty.Logger)
• type CMsgType
• type ClusterInToPub
• type ClusterInToPubShare
• type ClusterInToPubSys
• type PackageHandler
  • func (h *PackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error)
  • func (h *PackageHandler) Write(ss getty.Session, p interface{}) ([]byte, error)
• type WrapCMsg
  • func DecodeCMsg(b []byte) (WrapCMsg, int, error)
  • func NewWrapCMsgImpl(msgType CMsgType) WrapCMsg

Constants

const (
	CronPeriod      = 20e9
	WritePkgTimeout = 1e8
)

const (
	PubCMsg          CMsgType // 普通消息
	PubShareCMsg              // 共享主题消息
	PubShareRespCMsg          // 共享主题消息回应
	SubCMsg
	UnSubCMsg
	PingCMsg
	PingRespCMsg
	StatusCMsg   // 状态消息，可以通知其发慢点
	CloseSession // 通知集群删除某个session，断开某个client连接

	Tag byte = 0x01 // CMsgType中的tag字段为1时，第二个字节必须是此标志，第0位为1表示后面还有tag，为0表示后面没有tag了，为msg了
)

Variables

var (
	Cname = "name"
	Caddr = "addr"
)

var Sessions []getty.Session

Functions

func AddSession

func AddSession(name string, session getty.Session)

func ClientRequest

func ClientRequest()

func EncodeCMsg

func EncodeCMsg(msg WrapCMsg) ([]byte, error)

EncodeCMsg 将消息编码

func InitClusterTaskPool

func InitClusterTaskPool(poolSize int) (close io.Closer)

func NewEventListener

func NewEventListener() getty.EventListener

func NewServerEventListener

func NewServerEventListener() getty.EventListener

func RemoveSession

func RemoveSession(name string)

func SendMsgToCluster

func SendMsgToCluster(msg messagev5.Message, shareName, targetNode string, allSuccess func(message messagev5.Message),
	oneNodeSendSucFunc func(name string, message messagev5.Message),
	oneNodeSendFailFunc func(name string, message messagev5.Message))

SendMsgToCluster 发送消息到集群 shareName 共享主题组 targetNode 目标节点 这两个参数用于集群共享主题消息发送到特定的节点，TODO share msg 节点发送确认

func SetCurName

func SetCurName(listener getty.EventListener, curName string)

func SetLoggerLevelInfo

func SetLoggerLevelInfo()

func SetPubFunc

func SetPubFunc(el getty.EventListener, clusterInToPub ClusterInToPub,
	clusterInToPubShare ClusterInToPubShare, clusterInToPubSys ClusterInToPubSys,
	shareTopicMapNode cluster.ShareTopicMapNode)

func SetSessionOnOpen

func SetSessionOnOpen(listener getty.EventListener, fuc func(name string, session getty.Session))

func UpdateLogger

func UpdateLogger(lg getty.Logger)

更新getty内部日志，虽然都是zap

Types

type CMsgType

type CMsgType byte

CMsgType 前4位用来表示类型，和mqtt消息类型一样处理 剩余4bit，|....| 3 | 2 | 1 | 0 | 为1时分别表示

|共享主题消息，必有tag|     有无msg体    |  有无tag   |   为0   |
                        status不需要msg
                     ping和pingresp也可以没有

type ClusterInToPub

type ClusterInToPub func(msg1 *messagev5.PublishMessage) error

type ClusterInToPubShare

type ClusterInToPubShare func(msg1 *messagev5.PublishMessage, shareName string) error

type ClusterInToPubSys

type ClusterInToPubSys func(msg1 *messagev5.PublishMessage) error

type PackageHandler

type PackageHandler struct{}

func (*PackageHandler) Read

func (h *PackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error)

func (*PackageHandler) Write

func (h *PackageHandler) Write(ss getty.Session, p interface{}) ([]byte, error)

字节切片不会调用

type WrapCMsg

type WrapCMsg interface {
	Type() CMsgType
	Tag() []string
	Share() bool
	Msg() messagev5.Message
	Status() map[string]string // 状态数据，自定义
	CloseSessions() []string   // 需要断开连接并删除的session，client ids

	SetShare(shareName string, msg messagev5.Message)
	AddTag(tag string)
	SetMsg(msg messagev5.Message)
	Len() int
}

func DecodeCMsg

func DecodeCMsg(b []byte) (WrapCMsg, int, error)

DecodeCMsg 从消息体中解码包装的Msg，第二个返回值表示此处读取到的位置

func NewWrapCMsgImpl

func NewWrapCMsgImpl(msgType CMsgType) WrapCMsg