hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: complete cmpp platform phases 0-5

Browse Source
This commit is contained in:
hectorzhao
2026-07-01 13:22:04 +08:00
parent 824a8b334f
commit ee926fea04
86 changed files with 10490 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/README.md
+44
View File
@@ -0,0 +1,44 @@
# CMPP Gateway Spike
阶段 0 的 `gateway/` 目录用于 Go CMPP Gateway 技术 Spike。当前环境未安装 Go 工具链，先保留工程边界和验收清单。
## 阶段 0 职责
- 读取一个 CMPP 通道配置。
- 基于 gocmpp 或评估后的协议库完成 connect、submit、deliver、active test、terminate。
- 消费 `cmpp.submit.commands`。
- 发布 `cmpp.submit.results`、`cmpp.receipt.events`、`cmpp.uplink.events`。
- 维护 `messageId -> sequenceId -> gatewayMessageId` 映射。
- 支持断线重连和后续消息继续消费。
- 暴露健康检查和最小指标。
## 建议骨架
```text
gateway/
├── cmd/
│ ├── gateway/
│ └── smsc-simulator/
└── internal/
├── cmpp/
├── config/
├── connection/
├── metrics/
├── queue/
└── tracker/
```
## 当前环境状态
已安装 Go 1.26.4。当前 Spike 已完成：
- 队列消息结构定义。
- `messageId -> sequenceId -> gatewayMessageId` 追踪器。
- 内存模拟 submit resp 与 deliver 回执链路。
- 15000 条内存链路压测基线。
- gocmpp 编译级接入点。
- gocmpp 本地 TCP connect、submit、submit resp、active test 集成测试。
- gocmpp deliver 回执 PDU pack/unpack 测试。
- 断线重连状态机测试。
阶段 0 决策：协议层优先直接依赖 gocmpp，服务层连接管理、重连、SEQID/MSGID 追踪、队列、限速、监控、幂等由本项目自研。
gateway/cmd/gateway/main.go
+21
View File
@@ -0,0 +1,21 @@
package main
import (
"log"
"net/http"
"os"
"cmpp-platform/gateway/internal/health"
)
func main() {
addr := os.Getenv("GATEWAY_HEALTH_ADDR")
if addr == "" {
addr = ":8090"
}
log.Printf("cmpp gateway health server listening on %s", addr)
if err := http.ListenAndServe(addr, health.Handler()); err != nil {
log.Fatalf("gateway health server stopped: %v", err)
}
}
gateway/cmd/spike/main.go
+29
View File
@@ -0,0 +1,29 @@
package main
import (
"context"
"fmt"
"log"
"time"
"cmpp-platform/gateway/internal/spike"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
sim := spike.NewSimulator()
result, err := sim.RunLoad(ctx, 15000)
if err != nil {
log.Fatalf("run spike load: %v", err)
}
fmt.Printf("submitted=%d submitResults=%d receiptEvents=%d duration=%s throughput=%.2f msg/s\n",
result.Submitted,
result.SubmitResults,
result.ReceiptEvents,
result.Duration,
result.MessagesPerSec,
)
}
gateway/go.mod
+8
View File
@@ -0,0 +1,8 @@
module cmpp-platform/gateway
go 1.26
require (
github.com/bigwhite/gocmpp v0.0.0-20240917054108-b238366bff0b // indirect
golang.org/x/text v0.3.8 // indirect
)
gateway/go.sum
+28
View File
@@ -0,0 +1,28 @@
github.com/bigwhite/gocmpp v0.0.0-20240917054108-b238366bff0b h1:HOIU4bq4fpwWdtkpXASXnWk/fiFjd6o27Q0Kw4aJJAk=
github.com/bigwhite/gocmpp v0.0.0-20240917054108-b238366bff0b/go.mod h1:BDWS0X/2jJROFh0iYgdcAdv4jy3cPhVcZXvEkZmoqCM=
github.com/dvyukov/go-fuzz v0.0.0-20190516070045-5cc3605ccbb6/go.mod h1:11Gm+ccJnvAhCNLlf5+cS9KjtbaD5I5zaZpFMsTHWTw=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY=
golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gateway/internal/cmpp/client.go
+41
View File
@@ -0,0 +1,41 @@
package cmppadapter
import (
"time"
cmpp "github.com/bigwhite/gocmpp"
)
type Version string
const (
Version20 Version = "2.0"
Version30 Version = "3.0"
)
type ClientConfig struct {
Version Version
Address string
User string
Password string
Timeout time.Duration
}
type Client struct {
raw *cmpp.Client
}
func NewClient(config ClientConfig) *Client {
return &Client{raw: cmpp.NewClient(toProtocolType(config.Version))}
}
func (c *Client) Raw() *cmpp.Client {
return c.raw
}
func toProtocolType(version Version) cmpp.Type {
if version == Version20 {
return cmpp.V20
}
return cmpp.V30
}
gateway/internal/cmpp/client_test.go
+16
View File
@@ -0,0 +1,16 @@
package cmppadapter
import "testing"
func TestNewClientUsesGocmpp(t *testing.T) {
client := NewClient(ClientConfig{Version: Version30})
if client.Raw() == nil {
t.Fatal("expected gocmpp client")
}
}
func TestProtocolVersionMapping(t *testing.T) {
if toProtocolType(Version20).String() == toProtocolType(Version30).String() {
t.Fatal("expected CMPP 2.0 and 3.0 to map to different protocol types")
}
}
gateway/internal/cmpp/gocmpp_integration_test.go
+214
View File
@@ -0,0 +1,214 @@
package cmppadapter
import (
"bytes"
"crypto/md5"
"fmt"
"log"
"net"
"testing"
"time"
cmpp "github.com/bigwhite/gocmpp"
cmpputils "github.com/bigwhite/gocmpp/utils"
)
const (
spikeUser = "900001"
spikePassword = "888888"
)
func TestGocmppConnectSubmitAndActiveTest(t *testing.T) {
addr := reserveTCPAddr(t)
handlers := []cmpp.Handler{
cmpp.HandlerFunc(handleSpikeLogin),
cmpp.HandlerFunc(handleSpikeSubmit),
}
go func() {
err := cmpp.ListenAndServe(addr, cmpp.V30, 2*time.Second, 3, nil, handlers...)
if err != nil {
log.Printf("gocmpp spike server stopped: %v", err)
}
}()
time.Sleep(300 * time.Millisecond)
client := cmpp.NewClient(cmpp.V30)
defer client.Disconnect()
if err := client.Connect(addr, spikeUser, spikePassword, 2*time.Second); err != nil {
t.Fatalf("connect gocmpp server: %v", err)
}
content, err := cmpputils.Utf8ToUcs2("测试gocmpp submit")
if err != nil {
t.Fatalf("encode submit content: %v", err)
}
_, err = client.SendReqPkt(&cmpp.Cmpp3SubmitReqPkt{
PkTotal: 1,
PkNumber: 1,
RegisteredDelivery: 1,
MsgLevel: 1,
ServiceId: "test",
FeeUserType: 2,
FeeTerminalId: "13500002696",
FeeTerminalType: 0,
MsgFmt: 8,
MsgSrc: spikeUser,
FeeType: "02",
FeeCode: "10",
ValidTime: "151105131555101+",
AtTime: "",
SrcId: spikeUser,
DestUsrTl: 1,
DestTerminalId: []string{"13500002696"},
DestTerminalType: 0,
MsgLength: uint8(len(content)),
MsgContent: content,
})
if err != nil {
t.Fatalf("send submit: %v", err)
}
submitRsp := recvUntil[*cmpp.Cmpp3SubmitRspPkt](t, client, 2*time.Second)
if submitRsp.Result != 0 {
t.Fatalf("unexpected submit result: %d", submitRsp.Result)
}
if submitRsp.MsgId == 0 {
t.Fatal("expected gateway msg id")
}
_, err = client.SendReqPkt(&cmpp.CmppActiveTestReqPkt{})
if err != nil {
t.Fatalf("send active test: %v", err)
}
_ = recvUntil[*cmpp.CmppActiveTestRspPkt](t, client, 2*time.Second)
}
func TestGocmppDeliverReceiptPackAndUnpack(t *testing.T) {
receipt := &cmpp.CmppReceiptPkt{
MsgId: 12878564852733378560,
Stat: "DELIVRD",
SubmitTime: "2607010900",
DoneTime: "2607010901",
DestTerminalId: "13500002696",
SmscSequence: 42,
}
receiptBytes, err := receipt.Pack()
if err != nil {
t.Fatalf("pack receipt: %v", err)
}
deliver := &cmpp.Cmpp3DeliverReqPkt{
MsgId: 12878564852733378560,
DestId: "106900000000",
ServiceId: "test",
TpPid: 0,
TpUdhi: 0,
MsgFmt: 0,
SrcTerminalId: "13500002696",
SrcTerminalType: 0,
RegisterDelivery: 1,
MsgLength: uint8(cmpp.CmppReceiptPktLen),
MsgContent: string(receiptBytes),
}
data, err := deliver.Pack(1001)
if err != nil {
t.Fatalf("pack deliver receipt: %v", err)
}
var unpacked cmpp.Cmpp3DeliverReqPkt
if err := unpacked.Unpack(data[8:]); err != nil {
t.Fatalf("unpack deliver receipt: %v", err)
}
var gotReceipt cmpp.CmppReceiptPkt
if err := gotReceipt.Unpack([]byte(unpacked.MsgContent)); err != nil {
t.Fatalf("unpack receipt content: %v", err)
}
if gotReceipt.Stat != "DELIVRD" || gotReceipt.SmscSequence != 42 {
t.Fatalf("unexpected receipt payload: %+v", &gotReceipt)
}
}
func reserveTCPAddr(t *testing.T) string {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("reserve tcp addr: %v", err)
}
addr := listener.Addr().String()
if err := listener.Close(); err != nil {
t.Fatalf("close reserved listener: %v", err)
}
return addr
}
func recvUntil[T any](t *testing.T, client *cmpp.Client, timeout time.Duration) T {
t.Helper()
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
packet, err := client.RecvAndUnpackPkt(200 * time.Millisecond)
if err != nil {
continue
}
if typed, ok := packet.(T); ok {
return typed
}
}
var zero T
t.Fatalf("timed out waiting for %T", zero)
return zero
}
func handleSpikeLogin(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
req, ok := packet.Packer.(*cmpp.CmppConnReqPkt)
if !ok {
return true, nil
}
resp := response.Packer.(*cmpp.Cmpp3ConnRspPkt)
resp.Version = 0x30
if req.SrcAddr != cmpputils.OctetString(spikeUser, 6) {
resp.Status = uint32(cmpp.ErrnoConnInvalidSrcAddr)
return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnInvalidSrcAddr]
}
authSrc := md5.Sum(bytes.Join([][]byte{
[]byte(cmpputils.OctetString(spikeUser, 6)),
make([]byte, 9),
[]byte(spikePassword),
[]byte(cmpputils.TimeStamp2Str(req.Timestamp)),
}, nil))
if req.AuthSrc != string(authSrc[:]) {
resp.Status = uint32(cmpp.ErrnoConnAuthFailed)
return false, cmpp.ConnRspStatusErrMap[cmpp.ErrnoConnAuthFailed]
}
authIsmg := md5.Sum(bytes.Join([][]byte{{byte(resp.Status)}, authSrc[:], []byte(spikePassword)}, nil))
resp.AuthIsmg = string(authIsmg[:])
return false, nil
}
func handleSpikeSubmit(response *cmpp.Response, packet *cmpp.Packet, logger *log.Logger) (bool, error) {
req, ok := packet.Packer.(*cmpp.Cmpp3SubmitReqPkt)
if !ok {
return true, nil
}
if req.DestUsrTl == 0 || len(req.DestTerminalId) == 0 {
return false, fmt.Errorf("missing submit destination")
}
resp := response.Packer.(*cmpp.Cmpp3SubmitRspPkt)
resp.MsgId = 12878564852733378560
resp.Result = 0
return false, nil
}
gateway/internal/connection/reconnector.go
+44
View File
@@ -0,0 +1,44 @@
package connection
import (
"context"
"time"
)
type DialFunc func(context.Context) error
type Reconnector struct {
MaxAttempts int
Delay time.Duration
Dial DialFunc
}
func (r Reconnector) Connect(ctx context.Context) (int, error) {
attempts := r.MaxAttempts
if attempts <= 0 {
attempts = 1
}
var lastErr error
for attempt := 1; attempt <= attempts; attempt++ {
if err := ctx.Err(); err != nil {
return attempt - 1, err
}
if err := r.Dial(ctx); err != nil {
lastErr = err
if attempt < attempts && r.Delay > 0 {
select {
case <-ctx.Done():
return attempt, ctx.Err()
case <-time.After(r.Delay):
}
}
continue
}
return attempt, nil
}
return attempts, lastErr
}
gateway/internal/connection/reconnector_test.go
+47
View File
@@ -0,0 +1,47 @@
package connection
import (
"context"
"errors"
"testing"
)
func TestReconnectorRetriesAfterDisconnect(t *testing.T) {
failures := 0
reconnector := Reconnector{
MaxAttempts: 3,
Dial: func(context.Context) error {
failures++
if failures < 2 {
return errors.New("simulated disconnect")
}
return nil
},
}
attempts, err := reconnector.Connect(context.Background())
if err != nil {
t.Fatalf("connect after retry: %v", err)
}
if attempts != 2 {
t.Fatalf("expected success on second attempt, got %d", attempts)
}
}
func TestReconnectorReturnsLastError(t *testing.T) {
expected := errors.New("still disconnected")
reconnector := Reconnector{
MaxAttempts: 2,
Dial: func(context.Context) error {
return expected
},
}
attempts, err := reconnector.Connect(context.Background())
if !errors.Is(err, expected) {
t.Fatalf("expected last error, got %v", err)
}
if attempts != 2 {
t.Fatalf("expected two attempts, got %d", attempts)
}
}
gateway/internal/health/server.go
+26
View File
@@ -0,0 +1,26 @@
package health
import (
"encoding/json"
"net/http"
"time"
)
type Status struct {
Status string `json:"status"`
Service string `json:"service"`
Timestamp string `json:"timestamp"`
}
func Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(Status{
Status: "ok",
Service: "cmpp-gateway",
Timestamp: time.Now().UTC().Format(time.RFC3339Nano),
})
})
return mux
}
gateway/internal/health/server_test.go
+31
View File
@@ -0,0 +1,31 @@
package health
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
func TestHealthHandler(t *testing.T) {
server := httptest.NewServer(Handler())
defer server.Close()
response, err := http.Get(server.URL + "/health")
if err != nil {
t.Fatalf("get health: %v", err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK {
t.Fatalf("unexpected status: %d", response.StatusCode)
}
var payload Status
if err := json.NewDecoder(response.Body).Decode(&payload); err != nil {
t.Fatalf("decode health payload: %v", err)
}
if payload.Status != "ok" || payload.Service != "cmpp-gateway" {
t.Fatalf("unexpected payload: %+v", payload)
}
}
gateway/internal/queue/messages.go
+90
View File
@@ -0,0 +1,90 @@
package queue
import "time"
const SchemaVersion = "v1"
type MessageType string
const (
MessageTypeSubmitCommand MessageType = "SubmitCommand"
MessageTypeSubmitResult MessageType = "SubmitResult"
MessageTypeReceiptEvent MessageType = "ReceiptEvent"
MessageTypeUplinkEvent MessageType = "UplinkEvent"
)
type Envelope struct {
SchemaVersion string `json:"schemaVersion"`
MessageType MessageType `json:"messageType"`
TraceID string `json:"traceId"`
MessageID string `json:"messageId"`
ChannelID string `json:"channelId"`
CreatedAt time.Time `json:"createdAt"`
}
type SubmitCommand struct {
Envelope
TenantID string `json:"tenantId"`
ApplicationID string `json:"applicationId"`
TaskID string `json:"taskId,omitempty"`
SubmitID string `json:"submitId"`
PhoneNumber string `json:"phoneNumber"`
Content string `json:"content"`
Signature string `json:"signature"`
TemplateID string `json:"templateId"`
BillingUnits int `json:"billingUnits"`
Route Route `json:"route"`
CMPP CMPP `json:"cmpp"`
Retry Retry `json:"retry"`
}
type Route struct {
ChannelCode string `json:"channelCode"`
CMPPAccountCode string `json:"cmppAccountCode"`
Priority int `json:"priority"`
RateLimitPerSecond int `json:"rateLimitPerSecond,omitempty"`
}
type CMPP struct {
ServiceID string `json:"serviceId"`
SrcID string `json:"srcId"`
RegisteredDelivery int `json:"registeredDelivery"`
MsgFmt int `json:"msgFmt"`
FeeUserType int `json:"feeUserType,omitempty"`
FeeCode string `json:"feeCode,omitempty"`
FeeType string `json:"feeType,omitempty"`
}
type Retry struct {
Attempt int `json:"attempt"`
MaxAttempts int `json:"maxAttempts"`
}
type SubmitResult struct {
Envelope
SequenceID uint32 `json:"sequenceId"`
GatewayMessageID string `json:"gatewayMessageId"`
SubmitStatus string `json:"submitStatus"`
ErrorCode string `json:"errorCode,omitempty"`
ErrorMessage string `json:"errorMessage,omitempty"`
SubmittedAt time.Time `json:"submittedAt"`
}
type ReceiptEvent struct {
Envelope
SequenceID uint32 `json:"sequenceId"`
GatewayMessageID string `json:"gatewayMessageId"`
ReceiptStatus string `json:"receiptStatus"`
RawStatus string `json:"rawStatus"`
ErrorCode string `json:"errorCode,omitempty"`
DeliveredAt time.Time `json:"deliveredAt"`
}
type UplinkEvent struct {
Envelope
SequenceID uint32 `json:"sequenceId"`
PhoneNumber string `json:"phoneNumber"`
DestID string `json:"destId"`
Content string `json:"content"`
ReceivedAt time.Time `json:"receivedAt"`
}
gateway/internal/spike/simulator.go
+139
View File
@@ -0,0 +1,139 @@
package spike
import (
"context"
"fmt"
"sync/atomic"
"time"
"cmpp-platform/gateway/internal/queue"
"cmpp-platform/gateway/internal/tracker"
)
type Simulator struct {
tracker *tracker.Tracker
sequence atomic.Uint32
}
type RunResult struct {
Submitted int
SubmitResults int
ReceiptEvents int
Duration time.Duration
MessagesPerSec float64
}
func NewSimulator() *Simulator {
return &Simulator{tracker: tracker.New()}
}
func (s *Simulator) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, queue.ReceiptEvent, error) {
if err := ctx.Err(); err != nil {
return queue.SubmitResult{}, queue.ReceiptEvent{}, err
}
sequenceID := s.sequence.Add(1)
s.tracker.TrackSubmit(cmd.MessageID, sequenceID)
gatewayMessageID := fmt.Sprintf("gw-%s", cmd.MessageID)
mapping, err := s.tracker.TrackSubmitResp(sequenceID, gatewayMessageID)
if err != nil {
return queue.SubmitResult{}, queue.ReceiptEvent{}, err
}
now := time.Now().UTC()
result := queue.SubmitResult{
Envelope: queue.Envelope{
SchemaVersion: queue.SchemaVersion,
MessageType: queue.MessageTypeSubmitResult,
TraceID: cmd.TraceID,
MessageID: mapping.MessageID,
ChannelID: cmd.ChannelID,
CreatedAt: now,
},
SequenceID: mapping.SequenceID,
GatewayMessageID: mapping.GatewayMessageID,
SubmitStatus: "accepted",
SubmittedAt: now,
}
receipt := queue.ReceiptEvent{
Envelope: queue.Envelope{
SchemaVersion: queue.SchemaVersion,
MessageType: queue.MessageTypeReceiptEvent,
TraceID: cmd.TraceID,
MessageID: mapping.MessageID,
ChannelID: cmd.ChannelID,
CreatedAt: now.Add(10 * time.Millisecond),
},
SequenceID: mapping.SequenceID,
GatewayMessageID: mapping.GatewayMessageID,
ReceiptStatus: "delivered",
RawStatus: "DELIVRD",
DeliveredAt: now.Add(10 * time.Millisecond),
}
return result, receipt, nil
}
func (s *Simulator) RunLoad(ctx context.Context, count int) (RunResult, error) {
start := time.Now()
result := RunResult{Submitted: count}
for i := 0; i < count; i++ {
cmd := NewSubmitCommand(i)
_, _, err := s.Submit(ctx, cmd)
if err != nil {
return RunResult{}, err
}
result.SubmitResults++
result.ReceiptEvents++
}
result.Duration = time.Since(start)
if result.Duration > 0 {
result.MessagesPerSec = float64(count) / result.Duration.Seconds()
}
return result, nil
}
func NewSubmitCommand(index int) queue.SubmitCommand {
now := time.Now().UTC()
messageID := fmt.Sprintf("msg-spike-%06d", index)
return queue.SubmitCommand{
Envelope: queue.Envelope{
SchemaVersion: queue.SchemaVersion,
MessageType: queue.MessageTypeSubmitCommand,
TraceID: fmt.Sprintf("trace-spike-%06d", index),
MessageID: messageID,
ChannelID: "sms-channel-cmpp-spike",
CreatedAt: now,
},
TenantID: "tenant-spike",
ApplicationID: "app-spike",
TaskID: "task-spike",
SubmitID: fmt.Sprintf("submit-spike-%06d", index),
PhoneNumber: "13800138000",
Content: "您的验证码为 123456，5 分钟内有效。",
Signature: "测试平台",
TemplateID: "tpl-spike",
BillingUnits: 1,
Route: queue.Route{
ChannelCode: "CMCC-CMPP-SPIKE",
CMPPAccountCode: "cmpp-account-spike",
Priority: 10,
RateLimitPerSecond: 500,
},
CMPP: queue.CMPP{
ServiceID: "CMPP",
SrcID: "106900000000",
RegisteredDelivery: 1,
MsgFmt: 15,
FeeUserType: 2,
FeeCode: "0",
FeeType: "01",
},
Retry: queue.Retry{Attempt: 0, MaxAttempts: 3},
}
}
gateway/internal/spike/simulator_test.go
+48
View File
@@ -0,0 +1,48 @@
package spike
import (
"context"
"testing"
"time"
)
func TestSimulatorRunsOneMessageLifecycle(t *testing.T) {
sim := NewSimulator()
cmd := NewSubmitCommand(1)
result, receipt, err := sim.Submit(context.Background(), cmd)
if err != nil {
t.Fatalf("submit: %v", err)
}
if result.MessageID != cmd.MessageID {
t.Fatalf("submit result message id mismatch: %s != %s", result.MessageID, cmd.MessageID)
}
if result.SequenceID == 0 {
t.Fatal("expected sequence id")
}
if receipt.GatewayMessageID != result.GatewayMessageID {
t.Fatalf("receipt gateway id mismatch: %s != %s", receipt.GatewayMessageID, result.GatewayMessageID)
}
if receipt.ReceiptStatus != "delivered" {
t.Fatalf("unexpected receipt status: %s", receipt.ReceiptStatus)
}
}
func TestSimulatorLoadMeets500TPSFloor(t *testing.T) {
sim := NewSimulator()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
result, err := sim.RunLoad(ctx, 15000)
if err != nil {
t.Fatalf("run load: %v", err)
}
if result.SubmitResults != 15000 || result.ReceiptEvents != 15000 {
t.Fatalf("unexpected result counts: %+v", result)
}
if result.MessagesPerSec < 500 {
t.Fatalf("expected at least 500 msg/s, got %.2f", result.MessagesPerSec)
}
}
gateway/internal/tracker/tracker.go
+83
View File
@@ -0,0 +1,83 @@
package tracker
import (
"errors"
"sync"
)
var ErrMappingNotFound = errors.New("tracker mapping not found")
type Mapping struct {
MessageID string
SequenceID uint32
GatewayMessageID string
}
type Tracker struct {
mu sync.RWMutex
byMessage map[string]Mapping
bySeq map[uint32]string
byGateway map[string]string
}
func New() *Tracker {
return &Tracker{
byMessage: make(map[string]Mapping),
bySeq: make(map[uint32]string),
byGateway: make(map[string]string),
}
}
func (t *Tracker) TrackSubmit(messageID string, sequenceID uint32) Mapping {
t.mu.Lock()
defer t.mu.Unlock()
mapping := t.byMessage[messageID]
mapping.MessageID = messageID
mapping.SequenceID = sequenceID
t.byMessage[messageID] = mapping
t.bySeq[sequenceID] = messageID
return mapping
}
func (t *Tracker) TrackSubmitResp(sequenceID uint32, gatewayMessageID string) (Mapping, error) {
t.mu.Lock()
defer t.mu.Unlock()
messageID, ok := t.bySeq[sequenceID]
if !ok {
return Mapping{}, ErrMappingNotFound
}
mapping := t.byMessage[messageID]
mapping.GatewayMessageID = gatewayMessageID
t.byMessage[messageID] = mapping
t.byGateway[gatewayMessageID] = messageID
return mapping, nil
}
func (t *Tracker) ByGatewayMessageID(gatewayMessageID string) (Mapping, error) {
t.mu.RLock()
defer t.mu.RUnlock()
messageID, ok := t.byGateway[gatewayMessageID]
if !ok {
return Mapping{}, ErrMappingNotFound
}
return t.byMessage[messageID], nil
}
func (t *Tracker) ByMessageID(messageID string) (Mapping, error) {
t.mu.RLock()
defer t.mu.RUnlock()
mapping, ok := t.byMessage[messageID]
if !ok {
return Mapping{}, ErrMappingNotFound
}
return mapping, nil
}
gateway/internal/tracker/tracker_test.go
+36
View File
@@ -0,0 +1,36 @@
package tracker
import "testing"
func TestTrackerMapsMessageSequenceAndGatewayIDs(t *testing.T) {
tr := New()
submit := tr.TrackSubmit("msg-1", 1001)
if submit.MessageID != "msg-1" || submit.SequenceID != 1001 {
t.Fatalf("unexpected submit mapping: %+v", submit)
}
resp, err := tr.TrackSubmitResp(1001, "gw-1")
if err != nil {
t.Fatalf("track submit resp: %v", err)
}
if resp.GatewayMessageID != "gw-1" {
t.Fatalf("unexpected gateway message id: %+v", resp)
}
byGateway, err := tr.ByGatewayMessageID("gw-1")
if err != nil {
t.Fatalf("lookup by gateway id: %v", err)
}
if byGateway.MessageID != "msg-1" || byGateway.SequenceID != 1001 {
t.Fatalf("unexpected gateway lookup: %+v", byGateway)
}
}
func TestTrackerRejectsUnknownSubmitResp(t *testing.T) {
tr := New()
if _, err := tr.TrackSubmitResp(404, "gw-missing"); err == nil {
t.Fatal("expected missing mapping error")
}
}