Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Beta486 #491

Merged
merged 42 commits into from
Aug 20, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
42 commits
Select commit Hold shift + click to select a range
a6ebe57
beta447
Hoshinonyaruko Jun 29, 2024
1d63fae
beta448
Hoshinonyaruko Jul 1, 2024
cd34627
beta449
Hoshinonyaruko Jul 4, 2024
b790b29
beta450
Hoshinonyaruko Jul 4, 2024
7a9b56a
beta451
Hoshinonyaruko Jul 4, 2024
e17d0d4
beta452
Hoshinonyaruko Jul 4, 2024
1318175
beta453
Hoshinonyaruko Jul 4, 2024
b7f291c
beta454
Hoshinonyaruko Jul 4, 2024
f921895
beta455
Hoshinonyaruko Jul 9, 2024
24108c6
btea455
Hoshinonyaruko Jul 9, 2024
7cdc968
beta456
Hoshinonyaruko Jul 9, 2024
b5d784e
beta457
Hoshinonyaruko Jul 9, 2024
61e4001
beta458
Hoshinonyaruko Jul 9, 2024
6d4116a
beta460
Hoshinonyaruko Jul 9, 2024
2c1f1b7
beta460
Hoshinonyaruko Jul 11, 2024
5c312cd
beta461
Hoshinonyaruko Jul 12, 2024
58304eb
beta462
Hoshinonyaruko Jul 13, 2024
7e7c9f3
beta463
Hoshinonyaruko Jul 15, 2024
188f250
beta464
Hoshinonyaruko Jul 15, 2024
af0296d
beta465
Hoshinonyaruko Jul 18, 2024
500135c
beta467
Hoshinonyaruko Jul 21, 2024
c1a588c
beta468
Hoshinonyaruko Jul 21, 2024
1df8e15
beta469
Hoshinonyaruko Jul 21, 2024
f57ab7b
beta470
Hoshinonyaruko Jul 26, 2024
cd21ad9
beta471
Hoshinonyaruko Aug 1, 2024
6039272
beta472
Hoshinonyaruko Aug 1, 2024
c45b450
beta473
Hoshinonyaruko Aug 1, 2024
f89d6c5
beta473
Hoshinonyaruko Aug 3, 2024
b54fdee
beta475
Hoshinonyaruko Aug 5, 2024
a30a697
beta476
Hoshinonyaruko Aug 8, 2024
ab805cc
beta478
Hoshinonyaruko Aug 8, 2024
6debefd
beta479
Hoshinonyaruko Aug 17, 2024
9cc7e5d
beta479
Hoshinonyaruko Aug 17, 2024
c0fe93c
beta480
Hoshinonyaruko Aug 18, 2024
91cf223
beta481
Hoshinonyaruko Aug 18, 2024
a3aa6c1
beta482
Hoshinonyaruko Aug 18, 2024
0917e56
beta483
Hoshinonyaruko Aug 19, 2024
dd1f774
beta484
Hoshinonyaruko Aug 19, 2024
1c87d61
beta485
Hoshinonyaruko Aug 20, 2024
aac474a
beta486
Hoshinonyaruko Aug 20, 2024
2604eb7
beta486
Hoshinonyaruko Aug 20, 2024
9559b24
Merge branch 'main' into beta486
Hoshinonyaruko Aug 20, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
beta476
  • Loading branch information
Hoshinonyaruko committed Aug 8, 2024
commit a30a6973b233199b928bc5dd83bc5cfb91b52760
25 changes: 25 additions & 0 deletions botgo/event/event.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,34 @@ package event

import (
"encoding/json"
"sync"
"time"

"github.com/tidwall/gjson" // 由于回包的 d 类型不确定,gjson 用于从回包json中提取 d 并进行针对性的解析

"github.com/tencent-connect/botgo/dto"
)

func init() {
// Start a goroutine for periodic cleaning
go cleanProcessedIDs()
}

func cleanProcessedIDs() {
ticker := time.NewTicker(5 * time.Minute) // Adjust the interval as needed
defer ticker.Stop()

for range ticker.C {
// Clean processedIDs, remove entries which are no longer needed
processedIDs.Range(func(key, value interface{}) bool {
processedIDs.Delete(key)
return true
})
}
}

var processedIDs sync.Map

var eventParseFuncMap = map[dto.OPCode]map[dto.EventType]eventParseFunc{
dto.WSDispatchEvent: {
dto.EventGuildCreate: guildHandler,
Expand Down Expand Up @@ -218,6 +240,9 @@ func groupAtMessageHandler(payload *dto.WSPayload, message []byte) error {
if err := ParseData(message, data); err != nil {
return err
}
if _, loaded := processedIDs.LoadOrStore(data.ID, struct{}{}); loaded {
return nil
}
if DefaultHandlers.GroupATMessage != nil {
return DefaultHandlers.GroupATMessage(payload, data)
}
Expand Down
112 changes: 1 addition & 111 deletions botgo/websocket/client/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@ import (
"fmt"
"os"
"os/signal"
"sync"
"sync/atomic"
"syscall"
"time"
Expand All @@ -27,18 +26,6 @@ const DefaultQueueSize = 10000
// 定义全局变量
var global_s int64

// PayloadWithTimestamp 存储带时间戳的 WSPayload
type PayloadWithTimestamp struct {
Payload *dto.WSPayload
Timestamp time.Time
}

var dataMap sync.Map

func init() {
StartCleanupRoutine()
}

// Setup 依赖注册
func Setup() {
websocket.Register(&Client{})
Expand Down Expand Up @@ -200,33 +187,6 @@ func (c *Client) Session() *dto.Session {
return c.session
}

// func (c *Client) readMessageToQueue() {
// for {
// _, message, err := c.conn.ReadMessage()
// if err != nil {
// log.Errorf("%s read message failed, %v, message %s", c.session, err, string(message))
// close(c.messageQueue)
// c.closeChan <- err
// return
// }
// payload := &dto.WSPayload{}
// if err := json.Unmarshal(message, payload); err != nil {
// log.Errorf("%s json failed, %v", c.session, err)
// continue
// }
// // 更新 global_s 的值
// atomic.StoreInt64(&global_s, payload.S)

// payload.RawMessage = message
// log.Infof("%s receive %s message, %s", c.session, dto.OPMeans(payload.OPCode), string(message))
// // 处理内置的一些事件,如果处理成功,则这个事件不再投递给业务
// if c.isHandleBuildIn(payload) {
// continue
// }
// c.messageQueue <- payload
// }
// }

func (c *Client) readMessageToQueue() {
for {
_, message, err := c.conn.ReadMessage()
Expand All @@ -241,61 +201,19 @@ func (c *Client) readMessageToQueue() {
log.Errorf("%s json failed, %v", c.session, err)
continue
}
// 更新 global_s 的值
atomic.StoreInt64(&global_s, payload.S)

payload.RawMessage = message
log.Infof("%s receive %s message, %s", c.session, dto.OPMeans(payload.OPCode), string(message))

// 不过滤心跳事件
if payload.OPCode != 11 {
// 计算数据的哈希值
dataHash := calculateDataHash(payload.Data)

// 检查是否已存在相同的 Data
if existingPayload, ok := getDataFromSyncMap(dataHash); ok {
// 如果已存在相同的 Data,则丢弃当前消息
log.Infof("%s discard duplicate message with DataHash: %v", c.session, existingPayload)
continue
}

// 将新的 payload 存入 sync.Map
storeDataToSyncMap(dataHash, payload)
}

// 处理内置的一些事件,如果处理成功,则这个事件不再投递给业务
if c.isHandleBuildIn(payload) {
continue
}

c.messageQueue <- payload
}
}

func getDataFromSyncMap(dataHash string) (*dto.WSPayload, bool) {
value, ok := dataMap.Load(dataHash)
if !ok {
return nil, false
}
payloadWithTimestamp, ok := value.(*PayloadWithTimestamp)
if !ok {
return nil, false
}
return payloadWithTimestamp.Payload, true
}

func storeDataToSyncMap(dataHash string, payload *dto.WSPayload) {
payloadWithTimestamp := &PayloadWithTimestamp{
Payload: payload,
Timestamp: time.Now(),
}
dataMap.Store(dataHash, payloadWithTimestamp)
}

func calculateDataHash(data interface{}) string {
dataBytes, _ := json.Marshal(data)
return string(dataBytes) // 这里直接转换为字符串,可以使用更复杂的算法
}

// 在全局范围通过atomic访问s值与message_id的映射
func GetGlobalS() int64 {
return atomic.LoadInt64(&global_s)
Expand Down Expand Up @@ -379,31 +297,3 @@ func (c *Client) readyHandler(payload *dto.WSPayload) {
event.DefaultHandlers.Ready(payload, readyData)
}
}

const cleanupInterval = 5 * time.Minute // 清理间隔时间

func StartCleanupRoutine() {
go func() {
for {
<-time.After(cleanupInterval)
cleanupDataMap()
}
}()
}

func cleanupDataMap() {
now := time.Now()
dataMap.Range(func(key, value interface{}) bool {
payloadWithTimestamp, ok := value.(*PayloadWithTimestamp)
if !ok {
return true
}

// 检查时间戳,清理超过一定时间的数据
if now.Sub(payloadWithTimestamp.Timestamp) > cleanupInterval {
dataMap.Delete(key)
}

return true
})
}
Loading