channel

README

管理一个TCP连接的组件

Documentation

Index

• Variables
• func LogHeadByConnType(connType ConnType) string
• type AuthParams
• type Channel
  • func NewChannel(conf *conf.Config, conn *net.TCPConn, traceId int64, connType ConnType, ...) *Channel
  • func (c *Channel) CleanPath1()
  • func (c *Channel) CleanPath2()
  • func (c *Channel) CleanPath3()
  • func (c *Channel) Push(p *protocol.Proto) (err error)
  • func (c *Channel) SendFinish(logHead string)
  • func (c *Channel) SendReady()
  • func (c *Channel) SetWebSocketConnReaderWriter(wsConn *websocket.Conn)
  • func (c *Channel) Waiting() *protocol.Proto
• type CleanPath
• type ConnComponent
• type ConnType
• type Ring
  • func (r *Ring) AdvReadPointer()
  • func (r *Ring) AdvWritePointer()
  • func (r *Ring) GetProtoForRead() (proto *protocol.Proto, err error)
  • func (r *Ring) GetProtoForWrite() (proto *protocol.Proto, err error)
  • func (r *Ring) Init(max uint64)
  • func (r *Ring) ResetPointer()
• type Room
  • func NewRoom(id string) (r *Room)
  • func (r *Room) Close()
  • func (r *Room) DelChannel(ch *Channel)
  • func (r *Room) OnlineNum() int32
  • func (r *Room) PutChannel(ch *Channel) (err error)
  • func (r *Room) SendToAllChan(proto *protocol.Proto)
• type UserInfo

Variables

var (
	ErrRingEmpty = errors.New("ring buffer empty")
	ErrRingFull  = errors.New("ring buffer full")
)

Functions

func LogHeadByConnType

func LogHeadByConnType(connType ConnType) string

Types

type AuthParams

type AuthParams struct {
	UserInfo *UserInfo `json:"user_info"`
	Token    string    `json:"token"`
}

type Channel

type Channel struct {
	ConnComponent

	// the Room that Channel(User) is in
	Room *Room

	Next *Channel
	Prev *Channel

	// use bufio to reuse buffer
	Writer         *bufio.Writer
	Reader         *bufio.Reader
	ConnReadWriter protocol.ConnReadWriter

	// user info
	UserInfo *UserInfo

	// use to get proto，reduce GC
	ProtoAllocator Ring
	// contains filtered or unexported fields
}

Channel 每一个Channel代表一个长连接

func NewChannel

func NewChannel(conf *conf.Config, conn *net.TCPConn, traceId int64, connType ConnType, readerPool, writerPool *bytes.Pool, timerPool *newtimer.Timer) *Channel

NewChannel new a channel.

func (*Channel) CleanPath1

func (c *Channel) CleanPath1()

func (*Channel) CleanPath2

func (c *Channel) CleanPath2()

CleanPath2 Read协程结束时，需要执行的清理动作

func (*Channel) CleanPath3

func (c *Channel) CleanPath3()

CleanPath3 Write协程结束时，需要执行的清理动作

func (*Channel) Push

func (c *Channel) Push(p *protocol.Proto) (err error)

func (*Channel) SendFinish

func (c *Channel) SendFinish(logHead string)

SendFinish 通知Write协程结束

func (*Channel) SendReady

func (c *Channel) SendReady()

func (*Channel) SetWebSocketConnReaderWriter

func (c *Channel) SetWebSocketConnReaderWriter(wsConn *websocket.Conn)

func (*Channel) Waiting

func (c *Channel) Waiting() *protocol.Proto

type CleanPath

type CleanPath int

CleanPath 清理路径

const (
	CleanPath1 CleanPath = iota + 1
	CleanPath2
	CleanPath3
)

type ConnComponent

type ConnComponent struct {
	TraceId  int64
	ConnType ConnType

	// Connection(fd)
	Conn   *net.TCPConn
	WsConn *websocket.Conn

	// 分配buffer池子
	WriterPool *bytes.Pool
	ReaderPool *bytes.Pool

	// 分配定时器的池子
	TimerPool *newtimer.Timer
	// 从池子分配得到Timer
	Trd *newtimer.TimerData

	// 心跳相关
	LastHb     time.Time     // 上一次接收到心跳的时间
	HbExpire   time.Duration // 心跳超时的时间
	HbInterval time.Duration // 心跳续约频率
	// contains filtered or unexported fields
}

ConnComponent 每一条连接需要用到的组件

type ConnType

type ConnType int

const (
	ConnTypeTcp ConnType = iota + 1
	ConnTypeWebSocket
)

type Ring

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

Ring use to get proto，reduce GC Every User Has A Channel and Every Channel Has a Ring

func (*Ring) AdvReadPointer

func (r *Ring) AdvReadPointer()

AdvReadPointer 向前推进读指针

func (*Ring) AdvWritePointer

func (r *Ring) AdvWritePointer()

AdvWritePointer 向前推进写指针

func (*Ring) GetProtoForRead

func (r *Ring) GetProtoForRead() (proto *protocol.Proto, err error)

GetProtoForRead 获取一个Proto（用于读取的Proto） 后续通过 Ring.AdvReadPointer 推进读指针

func (*Ring) GetProtoForWrite

func (r *Ring) GetProtoForWrite() (proto *protocol.Proto, err error)

GetProtoForWrite 获取一个Proto（用于写入的Proto） 后续通过 Ring.AdvWritePointer 推进写指针

func (*Ring) Init

func (r *Ring) Init(max uint64)

func (*Ring) ResetPointer

func (r *Ring) ResetPointer()

ResetPointer 重置指针

type Room

type Room struct {
	Id string

	Online    int32 // 单台机器，单个房间的在线人数：dirty read is ok
	AllOnline int32 // 单台机器，所有房间的在线人数
	// contains filtered or unexported fields
}

Room is a room and store channel room info.

func NewRoom

func NewRoom(id string) (r *Room)

NewRoom new a room struct, store channel room info.

func (*Room) Close

func (r *Room) Close()

func (*Room) DelChannel

func (r *Room) DelChannel(ch *Channel)

DelChannel 从房间删除对象的Channel

func (*Room) OnlineNum

func (r *Room) OnlineNum() int32

OnlineNum 房间的在线人数

func (*Room) PutChannel

func (r *Room) PutChannel(ch *Channel) (err error)

PutChannel 把Channel放到房间中 insert to the head of the linklist

func (*Room) SendToAllChan

func (r *Room) SendToAllChan(proto *protocol.Proto)

SendToAllChan 把proto推送到房间中的所有Channel

type UserInfo

type UserInfo struct {
	TcpSessionId *tcp.SessionId `json:"tcp_session_id"` // 唯一地标识一条TCP连接
	RoomId       string         `json:"room_id"`        // 房间ID
	Platform     pb.Platform    `json:"platform"`       // 客户端平台
	IP           string         `json:"ip"`             // 客户端的IP地址
}