forked from zentures/surgemq
/
numtopic.go
82 lines (66 loc) · 1.58 KB
/
numtopic.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
package acl
import (
"fmt"
"sync"
)
type topicNumAuth struct {
topicTotalNowM sync.Map
topicUserM sync.Map
f GetAuthFunc
}
var _ Authenticator = (*topicNumAuth)(nil)
func (this *topicNumAuth) CheckPub(clientInfo *ClientInfo, topic string) bool {
return true
}
//需要区分tick和bar
func (this *topicNumAuth) CheckSub(clientInfo *ClientInfo, topic string) (success bool) {
if clientInfo == nil {
return false
}
defer func() {
log("SUB", topic, clientInfo)
}()
// fmt.Println("clientInfo",*clientInfo)
userName := clientInfo.GmToken
key := fmt.Sprintf(userTopicKeyFmt, userName, topic)
if _, ok := this.topicUserM.Load(key); ok {
success = true
return true
}
totalLimit := clientInfo.SubTopicLimit
totalNow, ok := this.topicTotalNowM.Load(userName)
if !ok {
this.topicTotalNowM.Store(userName, 1)
this.topicUserM.Store(key, true)
success = true
return
}
if totalNow.(int) >= totalLimit {
return
}
this.topicTotalNowM.Store(userName, totalNow.(int)+1)
this.topicUserM.Store(key, true)
success = true
return
}
func (this *topicNumAuth) ProcessUnSub(clientInfo *ClientInfo, topic string) {
if clientInfo == nil {
return
}
defer func() {
log("UNSUB", topic, clientInfo)
}()
userName := clientInfo.GmToken
key := fmt.Sprintf(userTopicKeyFmt, userName, topic)
if _, ok := this.topicUserM.Load(key); !ok {
return
}
this.topicUserM.Delete(key)
totalNow, ok := this.topicTotalNowM.Load(userName)
if ok {
this.topicTotalNowM.Store(userName, totalNow.(int)-1)
}
}
func (this *topicNumAuth) SetAuthFunc(f GetAuthFunc) {
this.f = f
}