sonet/pkg/protocol/session/channel.go

package session

import (
	"errors"
	"sonet/api/gen/postal"
	"sync"
)

var ErrChannelFullMsgDropped = errors.New("channel full, msg dropped")

// NetConn 各类型连接的 write 接口
type NetConn interface {
	Write([]byte) error
}

// Channel todo close channel
type Channel struct {
	Uid    string
	Conn   NetConn
	ch     chan *postal.Message
	groups []string // 记录 uid 对应的群组列表
	lock   *sync.Mutex
}

func NewChannel(uid string, conn NetConn) *Channel {
	return &Channel{
		Uid:  uid,
		Conn: conn,
		ch:   make(chan *postal.Message, 32),
		lock: &sync.Mutex{},
	}
}

func (c *Channel) Push(msg *postal.Message) (err error) {
	select {
	case c.ch <- msg:
	default:
		err = ErrChannelFullMsgDropped
	}
	return
}

func (c *Channel) Ready() *postal.Message {
	return <-c.ch
}

func (c *Channel) GroupJoin(gid string) {
	c.lock.Lock()
	defer c.lock.Unlock()
	for _, g := range c.groups {
		if g == gid { // 已加入
			return
		}
	}
	c.groups = append(c.groups, gid)
}

func (c *Channel) GroupLeave(gid string) {
	c.lock.Lock()
	defer c.lock.Unlock()

	for i := 0; i < len(c.groups); i++ {
		if c.groups[i] == gid { // 从i前移一位
			copy(c.groups[i:], c.groups[i+1:])
			c.groups = c.groups[0 : len(c.groups)-1]
			return
		}
	}
}

func (c *Channel) Groups() []string {
	return c.groups
}