pro/im_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完成鸽达达后台安全与审计改造

Browse Source
This commit is contained in:
richjohnrambo
2026-09-07 11:50:29 +08:00
parent ab2b84c9fa
commit bb87d01644
196 changed files with 19970 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/network/tcp/client.go
+2
View File
@@ -281,6 +281,8 @@ func (client *Client) handleRoutineTask(ctx context.Context, task func()) {
// Conn 获取当前连接
func (client *Client) Conn() *Conn {
client.mutex.Lock()
defer client.mutex.Unlock()
return client.conn
}
server/internal/library/network/tcp/rpc.go
+6 -2
View File
@@ -43,7 +43,6 @@ func (r *RPC) Request(ctx context.Context, msgId string, send func()) (res inter
defer func() {
isClose.Set(true)
close(resCh)
r.popCallback(msgId)
}()
@@ -57,8 +56,13 @@ func (r *RPC) Request(ctx context.Context, msgId string, send func()) (res inter
r.task(ctx, send)
timer := time.NewTimer(time.Second * RPCTimeout)
defer timer.Stop()
select {
case <-time.After(time.Second * RPCTimeout):
case <-ctx.Done():
err = gerror.Wrap(ctx.Err(), "RPC request canceled")
return
case <-timer.C:
err = gerror.New("RPC response timeout")
return
case got := <-resCh:
server/internal/library/network/tcp/tcp_example_test.go
+84 -97
View File
@@ -1,21 +1,14 @@
// Package tcp_test
// @Link https://github.com/bufanyun/hotgo
// @Copyright Copyright (c) 2023 HotGo CLI
// @Author Ms <133814250@qq.com>
// @License https://github.com/bufanyun/hotgo/blob/master/LICENSE
package tcp_test
import (
"context"
"fmt"
"github.com/gogf/gf/v2/os/gctx"
"github.com/gogf/gf/v2/test/gtest"
"hotgo/internal/library/network/tcp"
"net"
"testing"
"time"
)
var T *testing.T // 声明一个全局的 *testing.T 变量
"hotgo/internal/library/network/tcp"
)
type TestMsgReq struct {
Name string `json:"name"`
@@ -33,101 +26,95 @@ type TestRPCMsgRes struct {
tcp.ServerRes
}
func onTestMsg(ctx context.Context, req *TestMsgReq) {
fmt.Printf("服务器收到消息 ==> onTestMsg:%+v\n", req)
conn := tcp.ConnFromCtx(ctx)
gtest.C(T, func(t *gtest.T) {
t.AssertNE(conn, nil)
})
res := new(TestMsgRes)
res.Message = fmt.Sprintf("你的名字：%v", req.Name)
conn.Send(ctx, res)
}
func onResponseTestMsg(ctx context.Context, req *TestMsgRes) {
fmt.Printf("客户端收到响应消息 ==> TestMsgRes:%+v\n", req)
err := req.GetError()
gtest.C(T, func(t *gtest.T) {
t.AssertNil(err)
})
}
func onTestRPCMsg(ctx context.Context, req *TestRPCMsgReq) (res *TestRPCMsgRes, err error) {
fmt.Printf("服务器收到消息 ==> onTestRPCMsg:%+v\n", req)
res = new(TestRPCMsgRes)
res.Message = fmt.Sprintf("你的名字：%v", req.Name)
return
}
func startTCPServer() {
serv := tcp.NewServer(&tcp.ServerConfig{
Name: "hotgo",
Addr: ":8002",
})
// 注册路由
serv.RegisterRouter(
onTestMsg,
)
// 注册RPC路由
serv.RegisterRPCRouter(
onTestRPCMsg,
)
// 服务监听
err := serv.Listen()
gtest.C(T, func(t *gtest.T) {
t.AssertNil(err)
})
}
// 一个基本的消息收发
func TestSendMsg(t *testing.T) {
T = t
go startTCPServer()
ctx := gctx.New()
client := tcp.NewClient(&tcp.ClientConfig{
Addr: "127.0.0.1:8002",
address := freeTCPAddress(t)
messageResponse := make(chan string, 1)
server := tcp.NewServer(&tcp.ServerConfig{Name: "hotgo-test", Addr: address})
server.RegisterRouter(func(ctx context.Context, req *TestMsgReq) {
res := &TestMsgRes{}
res.SetMessage(fmt.Sprintf("你的名字：%s", req.Name))
if err := tcp.ConnFromCtx(ctx).Send(ctx, res); err != nil {
messageResponse <- "send-error:" + err.Error()
}
})
// 注册路由
client.RegisterRouter(
onResponseTestMsg,
)
server.RegisterRPCRouter(func(ctx context.Context, req *TestRPCMsgReq) (*TestRPCMsgRes, error) {
res := &TestRPCMsgRes{}
res.SetMessage(fmt.Sprintf("你的名字：%s", req.Name))
return res, nil
})
serverErrors := make(chan error, 1)
go func() {
err := client.Start()
gtest.C(T, func(t *gtest.T) {
t.AssertNil(err)
})
serverErrors <- server.Listen()
}()
defer server.Close()
// 确保服务都启动完成
time.Sleep(time.Second * 1)
// 拿到客户端的连接
conn := client.Conn()
gtest.C(T, func(t *gtest.T) {
t.AssertNE(conn, nil)
client := tcp.NewClient(&tcp.ClientConfig{
Addr: address,
Timeout: time.Second,
ConnectInterval: 10 * time.Millisecond,
MaxConnectCount: 100,
})
// 向服务器发送tcp消息，不会阻塞程序执行
err := conn.Send(ctx, &TestMsgReq{Name: "Tom"})
gtest.C(T, func(t *gtest.T) {
t.AssertNil(err)
client.RegisterRouter(func(ctx context.Context, res *TestMsgRes) {
messageResponse <- res.Message
})
if err := client.Start(); err != nil {
t.Fatalf("启动TCP客户端失败: %v", err)
}
defer client.Stop()
// 向服务器发送rpc消息，会等待服务器响应结果，直到拿到结果或响应超时才会继续
var res TestRPCMsgRes
if err = conn.RequestScan(ctx, &TestRPCMsgReq{Name: "Tony"}, &res); err != nil {
gtest.C(T, func(t *gtest.T) {
t.AssertNil(err)
})
conn := waitForConnection(t, client, serverErrors)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := conn.Send(ctx, &TestMsgReq{Name: "Tom"}); err != nil {
t.Fatalf("发送TCP消息失败: %v", err)
}
select {
case message := <-messageResponse:
if message != "你的名字：Tom" {
t.Fatalf("TCP响应不符: %s", message)
}
case <-ctx.Done():
t.Fatal("等待TCP响应超时")
}
fmt.Printf("客户端收到RPC消息响应 ==> TestRPCMsgRes:%+v\n", res)
time.Sleep(time.Second * 1)
var rpcResponse TestRPCMsgRes
if err := conn.RequestScan(ctx, &TestRPCMsgReq{Name: "Tony"}, &rpcResponse); err != nil {
t.Fatalf("发送RPC消息失败: %v", err)
}
if rpcResponse.Message != "你的名字：Tony" {
t.Fatalf("RPC响应不符: %s", rpcResponse.Message)
}
}
func freeTCPAddress(t *testing.T) string {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("申请测试端口失败: %v", err)
}
address := listener.Addr().String()
if err = listener.Close(); err != nil {
t.Fatalf("释放测试端口失败: %v", err)
}
return address
}
func waitForConnection(t *testing.T, client *tcp.Client, serverErrors <-chan error) *tcp.Conn {
t.Helper()
deadline := time.NewTimer(2 * time.Second)
defer deadline.Stop()
ticker := time.NewTicker(10 * time.Millisecond)
defer ticker.Stop()
for {
select {
case err := <-serverErrors:
t.Fatalf("TCP服务启动失败: %v", err)
case <-deadline.C:
t.Fatal("等待TCP连接超时")
case <-ticker.C:
if conn := client.Conn(); conn != nil {
return conn
}
}
}
}