diff --git a/cluster/gate/gate.go b/cluster/gate/gate.go
index 23c628af..dc641177 100644
--- a/cluster/gate/gate.go
+++ b/cluster/gate/gate.go
@@ -10,17 +10,15 @@ package gate
import (
"context"
"github.com/dobyte/due/cluster"
+ "github.com/dobyte/due/session"
"github.com/dobyte/due/transport"
- "sync"
"time"
- "github.com/dobyte/due/packet"
- "github.com/dobyte/due/registry"
- "github.com/dobyte/due/session"
-
"github.com/dobyte/due/component"
"github.com/dobyte/due/log"
"github.com/dobyte/due/network"
+ "github.com/dobyte/due/packet"
+ "github.com/dobyte/due/registry"
)
type Gate struct {
@@ -28,11 +26,10 @@ type Gate struct {
opts *options
ctx context.Context
cancel context.CancelFunc
- group *session.Group
- sessions sync.Pool
proxy *proxy
instance *registry.ServiceInstance
rpc transport.Server
+ session *session.Session
}
func NewGate(opts ...Option) *Gate {
@@ -43,9 +40,8 @@ func NewGate(opts ...Option) *Gate {
g := &Gate{}
g.opts = o
- g.group = session.NewGroup()
g.proxy = newProxy(g)
- g.sessions.New = func() interface{} { return session.NewSession() }
+ g.session = session.NewSession()
g.ctx, g.cancel = context.WithCancel(o.ctx)
return g
@@ -123,30 +119,21 @@ func (g *Gate) stopNetworkServer() {
// 处理连接打开
func (g *Gate) handleConnect(conn network.Conn) {
- s := g.sessions.Get().(*session.Session)
- s.Init(conn)
- g.group.AddSession(s)
+ g.session.AddConn(conn)
}
// 处理断开连接
func (g *Gate) handleDisconnect(conn network.Conn) {
- s, err := g.group.RemSession(session.Conn, conn.ID())
- if err != nil {
- log.Errorf("session remove failed, gid: %d, cid: %d, uid: %d, err: %v", g.opts.id, s.CID(), s.UID(), err)
- return
- }
+ g.session.RemConn(conn)
- if uid := conn.UID(); uid > 0 {
+ if cid, uid := conn.ID(), conn.UID(); uid > 0 {
ctx, cancel := context.WithTimeout(g.ctx, g.opts.timeout)
- err = g.proxy.unbindGate(ctx, conn.ID(), uid)
+ err := g.proxy.unbindGate(ctx, cid, uid)
cancel()
if err != nil {
log.Errorf("user unbind failed, gid: %d, uid: %d, err: %v", g.opts.id, uid, err)
}
}
-
- s.Reset()
- g.sessions.Put(s)
}
// 处理接收到的消息
@@ -157,8 +144,9 @@ func (g *Gate) handleReceive(conn network.Conn, data []byte, _ int) {
return
}
+ cid, uid := conn.ID(), conn.UID()
ctx, cancel := context.WithTimeout(g.ctx, g.opts.timeout)
- err = g.proxy.deliver(ctx, conn.ID(), conn.UID(), message)
+ err = g.proxy.deliver(ctx, cid, uid, message)
cancel()
if err != nil {
log.Warnf("deliver message failed: %v", err)
diff --git a/cluster/gate/provider.go b/cluster/gate/provider.go
index 920fe6a4..27523eb2 100644
--- a/cluster/gate/provider.go
+++ b/cluster/gate/provider.go
@@ -16,19 +16,17 @@ func (p *provider) Bind(ctx context.Context, cid, uid int64) error {
return ErrInvalidArgument
}
- s, err := p.gate.group.GetSession(session.Conn, cid)
+ err := p.gate.session.Bind(cid, uid)
if err != nil {
return err
}
err = p.gate.proxy.bindGate(ctx, cid, uid)
if err != nil {
- return err
+ _, _ = p.gate.session.Unbind(uid)
}
- s.Bind(uid)
-
- return nil
+ return err
}
// Unbind 解绑用户与网关间的关系
@@ -37,29 +35,22 @@ func (p *provider) Unbind(ctx context.Context, uid int64) error {
return ErrInvalidArgument
}
- s, err := p.gate.group.GetSession(session.User, uid)
+ cid, err := p.gate.session.Unbind(uid)
if err != nil {
return err
}
- err = p.gate.proxy.unbindGate(ctx, s.CID(), uid)
+ err = p.gate.proxy.unbindGate(ctx, cid, uid)
if err != nil {
return err
}
- s.Unbind(uid)
-
return nil
}
// GetIP 获取客户端IP地址
func (p *provider) GetIP(kind session.Kind, target int64) (string, error) {
- s, err := p.gate.group.GetSession(kind, target)
- if err != nil {
- return "", err
- }
-
- return s.RemoteIP()
+ return p.gate.session.RemoteIP(kind, target)
}
// Push 发送消息
@@ -69,19 +60,21 @@ func (p *provider) Push(kind session.Kind, target int64, message *packet.Message
return err
}
- return p.gate.group.Push(kind, target, msg)
+ return p.gate.session.Push(kind, target, msg)
}
// Multicast 推送组播消息
func (p *provider) Multicast(kind session.Kind, targets []int64, message *packet.Message) (int64, error) {
+ if len(targets) == 0 {
+ return 0, nil
+ }
+
msg, err := packet.Pack(message)
if err != nil {
return 0, err
}
- total, err := p.gate.group.Multicast(kind, targets, msg)
-
- return int64(total), err
+ return p.gate.session.Multicast(kind, targets, msg)
}
// Broadcast 推送广播消息
@@ -91,17 +84,10 @@ func (p *provider) Broadcast(kind session.Kind, message *packet.Message) (int64,
return 0, err
}
- total, err := p.gate.group.Broadcast(kind, msg)
-
- return int64(total), err
+ return p.gate.session.Broadcast(kind, msg)
}
// Disconnect 断开连接
func (p *provider) Disconnect(kind session.Kind, target int64, isForce bool) error {
- s, err := p.gate.group.GetSession(kind, target)
- if err != nil {
- return err
- }
-
- return s.Close(isForce)
+ return p.gate.session.Close(kind, target, isForce)
}
diff --git a/eventbus/kafka/go.mod b/eventbus/kafka/go.mod
index 554ddc68..256ccff6 100644
--- a/eventbus/kafka/go.mod
+++ b/eventbus/kafka/go.mod
@@ -4,7 +4,7 @@ go 1.16
require (
github.com/Shopify/sarama v1.38.1
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
)
replace github.com/dobyte/due => ./../../
diff --git a/eventbus/nats/go.mod b/eventbus/nats/go.mod
index 2f50555f..d5fd5eb7 100644
--- a/eventbus/nats/go.mod
+++ b/eventbus/nats/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/eventbus/nats
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/nats-io/nats-server/v2 v2.9.14 // indirect
github.com/nats-io/nats.go v1.23.0
)
diff --git a/eventbus/redis/go.mod b/eventbus/redis/go.mod
index e7ad4574..f2769f6b 100644
--- a/eventbus/redis/go.mod
+++ b/eventbus/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/eventbus/redis
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/go-redis/redis/v8 v8.11.5
)
diff --git a/locate/redis/go.mod b/locate/redis/go.mod
index fff0d81f..fde86c65 100644
--- a/locate/redis/go.mod
+++ b/locate/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/locate/redis
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/go-redis/redis/v8 v8.11.5
github.com/jonboulle/clockwork v0.3.0 // indirect
golang.org/x/sync v0.1.0
diff --git a/log/aliyun/go.mod b/log/aliyun/go.mod
index cbbb17e2..7c4435b9 100644
--- a/log/aliyun/go.mod
+++ b/log/aliyun/go.mod
@@ -4,7 +4,7 @@ go 1.16
require (
github.com/aliyun/aliyun-log-go-sdk v0.1.37
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/go-kit/kit v0.12.0 // indirect
github.com/go-kit/log v0.2.1 // indirect
github.com/pierrec/lz4 v2.6.1+incompatible // indirect
diff --git a/log/logrus/go.mod b/log/logrus/go.mod
index 29258d58..4a8eb344 100644
--- a/log/logrus/go.mod
+++ b/log/logrus/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/log/logrus
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/sirupsen/logrus v1.9.0
)
diff --git a/log/tencent/go.mod b/log/tencent/go.mod
index fb3cec43..490d42f1 100644
--- a/log/tencent/go.mod
+++ b/log/tencent/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/log/tencent
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/tencentcloud/tencentcloud-cls-sdk-go v1.0.2
)
diff --git a/log/zap/go.mod b/log/zap/go.mod
index 97cdb005..b3a6079e 100644
--- a/log/zap/go.mod
+++ b/log/zap/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/log/zap
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
go.uber.org/zap v1.22.0
)
diff --git a/network/tcp/go.mod b/network/tcp/go.mod
index 1135b8db..e9c83486 100644
--- a/network/tcp/go.mod
+++ b/network/tcp/go.mod
@@ -2,6 +2,6 @@ module github.com/dobyte/due/network/tcp
go 1.16
-require github.com/dobyte/due v0.0.21
+require github.com/dobyte/due v0.0.22
replace github.com/dobyte/due => ../../
diff --git a/network/ws/go.mod b/network/ws/go.mod
index 841d2f9d..657b3605 100644
--- a/network/ws/go.mod
+++ b/network/ws/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/network/ws
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/gorilla/websocket v1.5.0
)
diff --git a/registry/consul/go.mod b/registry/consul/go.mod
index 16c50e19..36af21f9 100644
--- a/registry/consul/go.mod
+++ b/registry/consul/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/consul
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/hashicorp/consul/api v1.13.0
)
diff --git a/registry/etcd/go.mod b/registry/etcd/go.mod
index 9d2afaaf..a58b0dc0 100644
--- a/registry/etcd/go.mod
+++ b/registry/etcd/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/etcd
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
go.etcd.io/etcd/api/v3 v3.5.4
go.etcd.io/etcd/client/v3 v3.5.4
)
diff --git a/session/group.go b/session/group.go
deleted file mode 100644
index 42210722..00000000
--- a/session/group.go
+++ /dev/null
@@ -1,231 +0,0 @@
-/**
- * @Author: fuxiao
- * @Email: 576101059@qq.com
- * @Date: 2022/6/9 19:58
- * @Desc: TODO
- */
-
-package session
-
-import (
- "errors"
- "sync"
-)
-
-var (
- ErrNotFoundSession = errors.New("the session not found in the group")
- ErrInvalidSessionKind = errors.New("invalid session kind")
-)
-
-const (
- Conn Kind = iota + 1 // 连接SESSION
- User // 用户SESSION
-)
-
-type Kind int
-
-type Group struct {
- rw sync.RWMutex // 读写锁
- conns map[int64]*Session // 连接会话（连接ID -> *Session）
- users map[int64]*Session // 用户会话（用户ID -> *Session）
-}
-
-func NewGroup() *Group {
- return &Group{
- conns: make(map[int64]*Session),
- users: make(map[int64]*Session),
- }
-}
-
-// AddSession 添加会话
-func (g *Group) AddSession(sess *Session) {
- g.rw.Lock()
- defer g.rw.Unlock()
-
- g.conns[sess.CID()] = sess
- if uid := sess.UID(); uid > 0 {
- g.users[uid] = sess
- }
- sess.addToGroups(g)
-}
-
-// RemSession 移除会话
-func (g *Group) RemSession(kind Kind, target int64) (*Session, error) {
- g.rw.Lock()
- defer g.rw.Unlock()
-
- switch kind {
- case Conn:
- sess, ok := g.conns[target]
- if !ok {
- return nil, ErrNotFoundSession
- }
-
- if uid := sess.UID(); uid > 0 {
- delete(g.users, uid)
- }
- delete(g.conns, target)
- sess.remFromGroups(g)
- return sess, nil
- case User:
- sess, ok := g.users[target]
- if !ok {
- return nil, ErrNotFoundSession
- }
-
- delete(g.users, target)
- delete(g.conns, sess.CID())
- sess.remFromGroups(g)
- return sess, nil
- default:
- return nil, ErrInvalidSessionKind
- }
-}
-
-// GetSession 获取会话
-func (g *Group) GetSession(kind Kind, target int64) (*Session, error) {
- g.rw.RLock()
- defer g.rw.RUnlock()
-
- return g.getSession(kind, target)
-}
-
-// 获取会话
-func (g *Group) getSession(kind Kind, target int64) (*Session, error) {
- switch kind {
- case Conn:
- sess, ok := g.conns[target]
- if !ok {
- return nil, ErrNotFoundSession
- }
- return sess, nil
- case User:
- sess, ok := g.users[target]
- if !ok {
- return nil, ErrNotFoundSession
- }
- return sess, nil
- default:
- return nil, ErrInvalidSessionKind
- }
-}
-
-// Send 发送消息（同步）
-func (g *Group) Send(kind Kind, target int64, msg []byte, typ ...int) error {
- sess, err := g.GetSession(kind, target)
- if err != nil {
- return err
- }
-
- return sess.Send(msg, typ...)
-}
-
-// Push 推送消息（异步）
-func (g *Group) Push(kind Kind, target int64, msg []byte, msgType ...int) error {
- sess, err := g.GetSession(kind, target)
- if err != nil {
- return err
- }
-
- return sess.Push(msg, msgType...)
-}
-
-// Multicast 推送组播消息（异步）
-func (g *Group) Multicast(kind Kind, targets []int64, msg []byte, msgType ...int) (n int, err error) {
- g.rw.RLock()
- defer g.rw.RUnlock()
-
- var sessions map[int64]*Session
- switch kind {
- case Conn:
- sessions = g.conns
- case User:
- sessions = g.users
- default:
- err = ErrInvalidSessionKind
- return
- }
-
- for _, target := range targets {
- session, ok := sessions[target]
- if !ok {
- continue
- }
- if session.Push(msg, msgType...) == nil {
- n++
- }
- }
-
- return
-}
-
-// Broadcast 推送广播消息（异步）
-func (g *Group) Broadcast(kind Kind, msg []byte, msgType ...int) (n int, err error) {
- g.rw.RLock()
- defer g.rw.RUnlock()
-
- var sessions map[int64]*Session
- switch kind {
- case Conn:
- sessions = g.conns
- case User:
- sessions = g.users
- default:
- err = ErrInvalidSessionKind
- return
- }
-
- for _, session := range sessions {
- if session.Push(msg, msgType...) == nil {
- n++
- }
- }
-
- return
-}
-
-// 添加会话
-func (g *Group) addSession(sess *Session) {
- g.rw.Lock()
- defer g.rw.Unlock()
-
- g.conns[sess.CID()] = sess
- if uid := sess.UID(); uid > 0 {
- g.users[uid] = sess
- }
-}
-
-// 移除会话
-func (g *Group) remSession(cid int64) error {
- g.rw.Lock()
- defer g.rw.Unlock()
-
- sess, ok := g.conns[cid]
- if !ok {
- return ErrNotFoundSession
- }
-
- delete(g.conns, cid)
-
- if uid := sess.UID(); uid > 0 {
- delete(g.users, uid)
- }
-
- return nil
-}
-
-// 添加用户会话
-func (g *Group) addUserSession(uid int64, sess *Session) {
- g.rw.Lock()
- defer g.rw.Unlock()
-
- g.users[uid] = sess
-}
-
-// 移除用户会话
-func (g *Group) remUserSession(uid int64) {
- g.rw.Lock()
- defer g.rw.Unlock()
-
- delete(g.users, uid)
-}
diff --git a/session/session.go b/session/session.go
index abff6163..6d28477d 100644
--- a/session/session.go
+++ b/session/session.go
@@ -1,179 +1,278 @@
-/**
- * @Author: fuxiao
- * @Email: 576101059@qq.com
- * @Date: 2022/6/9 20:10
- * @Desc: TODO
- */
-
package session
import (
+ "github.com/dobyte/due/errors"
+ "github.com/dobyte/due/network"
"net"
"sync"
+)
- "github.com/dobyte/due/network"
+var (
+ ErrNotFoundSession = errors.New("not found session")
+ ErrInvalidSessionKind = errors.New("invalid session kind")
)
+const (
+ Conn Kind = iota + 1 // 连接SESSION
+ User // 用户SESSION
+)
+
+type Kind int
+
type Session struct {
- rw sync.RWMutex // 读写锁
- conn network.Conn // 连接
- groups map[*Group]struct{} // 所在组
+ rw sync.RWMutex // 读写锁
+ conns map[int64]network.Conn // 连接会话（连接ID -> network.Conn）
+ users map[int64]network.Conn // 用户会话（用户ID -> network.Conn）
}
func NewSession() *Session {
- return &Session{}
+ return &Session{
+ conns: make(map[int64]network.Conn),
+ users: make(map[int64]network.Conn),
+ }
}
-// Init 初始化会话
-func (s *Session) Init(conn network.Conn) {
+// AddConn 添加连接
+func (s *Session) AddConn(conn network.Conn) {
s.rw.Lock()
defer s.rw.Unlock()
- s.conn = conn
- s.groups = make(map[*Group]struct{})
-}
+ cid, uid := conn.ID(), conn.UID()
-// Reset 重置会话
-func (s *Session) Reset() {
- s.rw.Lock()
- defer s.rw.Unlock()
+ s.conns[cid] = conn
- s.groups = nil
+ if uid != 0 {
+ s.users[uid] = conn
+ }
}
-// CID 获取连接ID
-func (s *Session) CID() int64 {
- s.rw.RLock()
- defer s.rw.RUnlock()
+// RemConn 移除连接
+func (s *Session) RemConn(conn network.Conn) {
+ s.rw.Lock()
+ defer s.rw.Unlock()
- return s.conn.ID()
-}
+ cid, uid := conn.ID(), conn.UID()
-// UID 获取用户ID
-func (s *Session) UID() int64 {
- s.rw.RLock()
- defer s.rw.RUnlock()
+ delete(s.conns, cid)
- return s.conn.UID()
+ if uid != 0 {
+ delete(s.users, uid)
+ }
}
// Bind 绑定用户ID
-func (s *Session) Bind(uid int64) {
+func (s *Session) Bind(cid, uid int64) error {
s.rw.Lock()
defer s.rw.Unlock()
- s.conn.Bind(uid)
- for group := range s.groups {
- group.addUserSession(uid, s)
+ conn, err := s.conn(Conn, cid)
+ if err != nil {
+ return err
}
+
+ if oldUID := conn.UID(); oldUID != 0 {
+ delete(s.users, oldUID)
+ }
+
+ conn.Bind(uid)
+ s.users[uid] = conn
+
+ return nil
}
// Unbind 解绑用户ID
-func (s *Session) Unbind(uid int64) {
+func (s *Session) Unbind(uid int64) (int64, error) {
s.rw.Lock()
defer s.rw.Unlock()
- s.conn.Unbind()
- for group := range s.groups {
- group.remUserSession(uid)
+ conn, err := s.conn(User, uid)
+ if err != nil {
+ return 0, err
}
-}
-// Close 关闭会话
-func (s *Session) Close(isForce ...bool) error {
- s.rw.Lock()
- defer s.rw.Unlock()
+ conn.Unbind()
+ delete(s.users, uid)
- return s.conn.Close(isForce...)
+ return conn.ID(), nil
}
// LocalIP 获取本地IP
-func (s *Session) LocalIP() (string, error) {
+func (s *Session) LocalIP(kind Kind, target int64) (string, error) {
s.rw.RLock()
defer s.rw.RUnlock()
- return s.conn.LocalIP()
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return "", err
+ }
+
+ return conn.LocalIP()
}
// LocalAddr 获取本地地址
-func (s *Session) LocalAddr() (net.Addr, error) {
+func (s *Session) LocalAddr(kind Kind, target int64) (net.Addr, error) {
s.rw.RLock()
defer s.rw.RUnlock()
- return s.conn.LocalAddr()
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return nil, err
+ }
+
+ return conn.LocalAddr()
}
// RemoteIP 获取远端IP
-func (s *Session) RemoteIP() (string, error) {
+func (s *Session) RemoteIP(kind Kind, target int64) (string, error) {
s.rw.RLock()
defer s.rw.RUnlock()
- return s.conn.RemoteIP()
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return "", err
+ }
+
+ return conn.RemoteIP()
}
// RemoteAddr 获取远端地址
-func (s *Session) RemoteAddr() (net.Addr, error) {
+func (s *Session) RemoteAddr(kind Kind, target int64) (net.Addr, error) {
s.rw.RLock()
defer s.rw.RUnlock()
- return s.conn.RemoteAddr()
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return nil, err
+ }
+
+ return conn.RemoteAddr()
}
-// Send 发送消息（同步）
-func (s *Session) Send(msg []byte, msgType ...int) error {
+// Close 关闭会话
+func (s *Session) Close(kind Kind, target int64, isForce ...bool) error {
s.rw.RLock()
defer s.rw.RUnlock()
- return s.conn.Send(msg, msgType...)
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return err
+ }
+
+ cid, uid := conn.ID(), conn.UID()
+
+ err = conn.Close(isForce...)
+ if err != nil {
+ return err
+ }
+
+ delete(s.conns, cid)
+ if uid != 0 {
+ delete(s.users, uid)
+ }
+
+ return nil
}
-// Push 发送消息（异步）
-func (s *Session) Push(msg []byte, msgType ...int) error {
+// Send 发送消息（同步）
+func (s *Session) Send(kind Kind, target int64, msg []byte, msgType ...int) error {
s.rw.RLock()
defer s.rw.RUnlock()
- return s.conn.Push(msg, msgType...)
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return err
+ }
+
+ return conn.Send(msg, msgType...)
}
-// AddToGroups 添加到会话组
-func (s *Session) AddToGroups(groups ...*Group) {
- s.rw.Lock()
- defer s.rw.Unlock()
+// Push 推送消息（异步）
+func (s *Session) Push(kind Kind, target int64, msg []byte, msgType ...int) error {
+ s.rw.RLock()
+ defer s.rw.RUnlock()
- for i := range groups {
- group := groups[i]
- group.addSession(s)
- s.groups[group] = struct{}{}
+ conn, err := s.conn(kind, target)
+ if err != nil {
+ return err
}
+
+ return conn.Push(msg, msgType...)
}
-// 添加到会话组
-func (s *Session) addToGroups(groups ...*Group) {
- s.rw.Lock()
- defer s.rw.Unlock()
+// Multicast 推送组播消息（异步）
+func (s *Session) Multicast(kind Kind, targets []int64, msg []byte, msgType ...int) (n int64, err error) {
+ if len(targets) == 0 {
+ return
+ }
+
+ s.rw.RLock()
+ defer s.rw.RUnlock()
- for i := range groups {
- s.groups[groups[i]] = struct{}{}
+ var conns map[int64]network.Conn
+ switch kind {
+ case Conn:
+ conns = s.conns
+ case User:
+ conns = s.users
+ default:
+ err = ErrInvalidSessionKind
+ return
}
+
+ for _, target := range targets {
+ conn, ok := conns[target]
+ if !ok {
+ continue
+ }
+ if conn.Push(msg, msgType...) == nil {
+ n++
+ }
+ }
+
+ return
}
-// RemFromGroups 从会话组移除
-func (s *Session) RemFromGroups(groups ...*Group) {
- s.rw.Lock()
- defer s.rw.Unlock()
+// Broadcast 推送广播消息（异步）
+func (s *Session) Broadcast(kind Kind, msg []byte, msgType ...int) (n int64, err error) {
+ s.rw.RLock()
+ defer s.rw.RUnlock()
- for _, group := range groups {
- _ = group.remSession(s.CID())
- delete(s.groups, group)
+ var conns map[int64]network.Conn
+ switch kind {
+ case Conn:
+ conns = s.conns
+ case User:
+ conns = s.users
+ default:
+ err = ErrInvalidSessionKind
+ return
}
-}
-// 从会话组移除
-func (s *Session) remFromGroups(groups ...*Group) {
- s.rw.Lock()
- defer s.rw.Unlock()
+ for _, conn := range conns {
+ if conn.Push(msg, msgType...) == nil {
+ n++
+ }
+ }
+
+ return
+}
- for _, group := range groups {
- delete(s.groups, group)
+// 获取会话
+func (s *Session) conn(kind Kind, target int64) (network.Conn, error) {
+ switch kind {
+ case Conn:
+ conn, ok := s.conns[target]
+ if !ok {
+ return nil, ErrNotFoundSession
+ }
+ return conn, nil
+ case User:
+ conn, ok := s.users[target]
+ if !ok {
+ return nil, ErrNotFoundSession
+ }
+ return conn, nil
+ default:
+ return nil, ErrInvalidSessionKind
}
}
diff --git a/transport/grpc/go.mod b/transport/grpc/go.mod
index 73a7ef34..59f083c6 100644
--- a/transport/grpc/go.mod
+++ b/transport/grpc/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/transport/grpc
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/golang/protobuf v1.5.2
google.golang.org/grpc v1.50.1
google.golang.org/protobuf v1.28.1 // indirect
diff --git a/transport/rpcx/go.mod b/transport/rpcx/go.mod
index 9ada7f97..abc668ef 100644
--- a/transport/rpcx/go.mod
+++ b/transport/rpcx/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/transport/rpcx
go 1.16
require (
- github.com/dobyte/due v0.0.21
+ github.com/dobyte/due v0.0.22
github.com/smallnest/rpcx v1.7.11
)
diff --git a/version.go b/version.go
index b8028de3..2d82d44e 100644
--- a/version.go
+++ b/version.go
@@ -8,7 +8,7 @@
package due
// Version 框架版本
-const Version = "v0.0.21"
+const Version = "v0.0.22"
// Website 框架网址
const Website = "https://github.com/dobyte/due"