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

fix: submit cmpp2 packets for cmpp2 channels

Browse Source
This commit is contained in:
hectorzhao
2026-07-09 16:01:10 +08:00
parent 64bb3fb430
commit 20757203d9
3 changed files with 190 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/internal/upstream/manager.go
+101 -33
View File
@@ -390,8 +390,10 @@ type connection struct {
}
type submitPartResponse struct {
rsp *cmpp.Cmpp3SubmitRspPkt
err error
seqID uint32
msgID uint64
result uint32
err error
}
func (c *connection) matches(config queue.UpstreamConfig) bool {
@@ -419,25 +421,7 @@ func (c *connection) ensureConnected() (bool, error) {
func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) {
rspCh := make(chan submitPartResponse, 1)
pkt := &cmpp.Cmpp3SubmitReqPkt{
PkTotal: part.PkTotal,
PkNumber: part.PkNumber,
TpUdhi: part.TpUdhi,
RegisteredDelivery: uint8(cmd.CMPP.RegisteredDelivery),
MsgLevel: 1,
ServiceId: cmd.CMPP.ServiceID,
FeeUserType: uint8(defaultInt(cmd.CMPP.FeeUserType, 2)),
FeeTerminalId: cmd.PhoneNumber,
MsgFmt: uint8(cmd.CMPP.MsgFmt),
MsgSrc: c.config.Account,
FeeType: defaultString(cmd.CMPP.FeeType, "02"),
FeeCode: defaultString(cmd.CMPP.FeeCode, "0"),
SrcId: cmd.CMPP.SrcID,
DestUsrTl: 1,
DestTerminalId: []string{cmd.PhoneNumber},
MsgLength: uint8(len(part.MsgContent)),
MsgContent: part.MsgContent,
}
pkt := c.submitRequestPacket(cmd, part)
c.sendMu.Lock()
seq, err := c.client.SendReqPkt(pkt)
@@ -468,29 +452,106 @@ func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, pa
result := submitResult(cmd, seq, "", "timeout", "CONNECTION_LOST", rsp.err.Error())
return seq, "", result, rsp.err
}
if rsp.rsp == nil {
err := fmt.Errorf("submit response is empty")
result := submitResult(cmd, seq, "", "timeout", "EMPTY_SUBMIT_RESPONSE", err.Error())
return seq, "", result, err
}
gatewayMessageID := fmt.Sprint(rsp.rsp.MsgId)
gatewayMessageID := fmt.Sprint(rsp.msgID)
status := "accepted"
errorCode := ""
errorMessage := ""
if rsp.rsp.Result != 0 {
if rsp.result != 0 {
status = "rejected"
errorCode = fmt.Sprint(rsp.rsp.Result)
errorMessage = fmt.Sprintf("upstream submit rejected with result %d", rsp.rsp.Result)
errorCode = fmt.Sprint(rsp.result)
errorMessage = fmt.Sprintf("upstream submit rejected with result %d", rsp.result)
}
if rsp.rsp.Result == 0 {
if rsp.result == 0 {
c.mu.Lock()
c.tracker[rsp.rsp.MsgId] = cmd
c.tracker[rsp.msgID] = cmd
c.mu.Unlock()
}
return seq, gatewayMessageID, submitResult(cmd, seq, gatewayMessageID, status, errorCode, errorMessage), nil
}
}
func (c *connection) submitRequestPacket(cmd queue.SubmitCommand, part submitPart) cmpp.Packer {
base := submitRequestFields{
PkTotal: part.PkTotal,
PkNumber: part.PkNumber,
TpUdhi: part.TpUdhi,
RegisteredDelivery: uint8(cmd.CMPP.RegisteredDelivery),
MsgLevel: 1,
ServiceId: cmd.CMPP.ServiceID,
FeeUserType: uint8(defaultInt(cmd.CMPP.FeeUserType, 2)),
FeeTerminalId: cmd.PhoneNumber,
MsgFmt: uint8(cmd.CMPP.MsgFmt),
MsgSrc: c.config.Account,
FeeType: defaultString(cmd.CMPP.FeeType, "02"),
FeeCode: defaultString(cmd.CMPP.FeeCode, "0"),
SrcId: cmd.CMPP.SrcID,
DestUsrTl: 1,
DestTerminalId: []string{cmd.PhoneNumber},
MsgLength: uint8(len(part.MsgContent)),
MsgContent: part.MsgContent,
}
if protocolVersion(c.config.CMPPVersion) == cmpp.V20 {
return &cmpp.Cmpp2SubmitReqPkt{
PkTotal: base.PkTotal,
PkNumber: base.PkNumber,
RegisteredDelivery: base.RegisteredDelivery,
MsgLevel: base.MsgLevel,
ServiceId: base.ServiceId,
FeeUserType: base.FeeUserType,
FeeTerminalId: base.FeeTerminalId,
TpUdhi: base.TpUdhi,
MsgFmt: base.MsgFmt,
MsgSrc: base.MsgSrc,
FeeType: base.FeeType,
FeeCode: base.FeeCode,
SrcId: base.SrcId,
DestUsrTl: base.DestUsrTl,
DestTerminalId: base.DestTerminalId,
MsgLength: base.MsgLength,
MsgContent: base.MsgContent,
}
}
return &cmpp.Cmpp3SubmitReqPkt{
PkTotal: base.PkTotal,
PkNumber: base.PkNumber,
RegisteredDelivery: base.RegisteredDelivery,
MsgLevel: base.MsgLevel,
ServiceId: base.ServiceId,
FeeUserType: base.FeeUserType,
FeeTerminalId: base.FeeTerminalId,
TpUdhi: base.TpUdhi,
MsgFmt: base.MsgFmt,
MsgSrc: base.MsgSrc,
FeeType: base.FeeType,
FeeCode: base.FeeCode,
SrcId: base.SrcId,
DestUsrTl: base.DestUsrTl,
DestTerminalId: base.DestTerminalId,
MsgLength: base.MsgLength,
MsgContent: base.MsgContent,
}
}
type submitRequestFields struct {
PkTotal uint8
PkNumber uint8
RegisteredDelivery uint8
MsgLevel uint8
ServiceId string
FeeUserType uint8
FeeTerminalId string
TpUdhi uint8
MsgFmt uint8
MsgSrc string
FeeType string
FeeCode string
SrcId string
DestUsrTl uint8
DestTerminalId []string
MsgLength uint8
MsgContent string
}
func (c *connection) tryAcquireWindow() bool {
if c.window == nil {
c.window = make(chan struct{}, defaultWindowSize)
@@ -530,12 +591,19 @@ func (c *connection) readLoop() {
continue
}
switch p := pkt.(type) {
case *cmpp.Cmpp2SubmitRspPkt:
c.mu.Lock()
ch := c.pending[p.SeqId]
c.mu.Unlock()
if ch != nil {
ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: uint32(p.Result)}
}
case *cmpp.Cmpp3SubmitRspPkt:
c.mu.Lock()
ch := c.pending[p.SeqId]
c.mu.Unlock()
if ch != nil {
ch <- submitPartResponse{rsp: p}
ch <- submitPartResponse{seqID: p.SeqId, msgID: p.MsgId, result: p.Result}
}
case *cmpp.Cmpp3DeliverReqPkt:
c.handleDeliver(p)
gateway/internal/upstream/submit_packet_test.go
+88
View File
@@ -0,0 +1,88 @@
package upstream
import (
"testing"
"time"
"cmpp-platform/gateway/internal/queue"
cmpp "github.com/bigwhite/gocmpp"
)
func TestSubmitRequestPacketUsesCMPP2PacketForCMPP20Channel(t *testing.T) {
conn := &connection{config: queue.UpstreamConfig{CMPPVersion: "2.0", Account: "ljcs02"}}
cmd := submitCommandForPacketTest("2.0")
parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
if err != nil {
t.Fatalf("splitSubmitContent returned error: %v", err)
}
packet, ok := conn.submitRequestPacket(cmd, parts[0]).(*cmpp.Cmpp2SubmitReqPkt)
if !ok {
t.Fatalf("expected Cmpp2SubmitReqPkt, got %T", packet)
}
if packet.MsgFmt != 8 {
t.Fatalf("MsgFmt = %d, want 8", packet.MsgFmt)
}
if packet.MsgSrc != "ljcs02" {
t.Fatalf("MsgSrc = %q, want ljcs02", packet.MsgSrc)
}
if packet.SrcId != "1069999999" {
t.Fatalf("SrcId = %q, want 1069999999", packet.SrcId)
}
if packet.MsgLength != uint8(len(parts[0].MsgContent)) {
t.Fatalf("MsgLength = %d, want %d", packet.MsgLength, len(parts[0].MsgContent))
}
if len(packet.MsgContent)%2 != 0 {
t.Fatalf("CMPP2 UCS2 payload length must be even, got %d", len(packet.MsgContent))
}
if packet.DestTerminalId[0] != cmd.PhoneNumber {
t.Fatalf("DestTerminalId = %q, want %q", packet.DestTerminalId[0], cmd.PhoneNumber)
}
}
func TestSubmitRequestPacketUsesCMPP3PacketForCMPP30Channel(t *testing.T) {
conn := &connection{config: queue.UpstreamConfig{CMPPVersion: "3.0", Account: "ljcs02"}}
cmd := submitCommandForPacketTest("3.0")
parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
if err != nil {
t.Fatalf("splitSubmitContent returned error: %v", err)
}
packet, ok := conn.submitRequestPacket(cmd, parts[0]).(*cmpp.Cmpp3SubmitReqPkt)
if !ok {
t.Fatalf("expected Cmpp3SubmitReqPkt, got %T", packet)
}
if packet.MsgFmt != 8 {
t.Fatalf("MsgFmt = %d, want 8", packet.MsgFmt)
}
if packet.MsgLength != uint8(len(parts[0].MsgContent)) {
t.Fatalf("MsgLength = %d, want %d", packet.MsgLength, len(parts[0].MsgContent))
}
}
func submitCommandForPacketTest(cmppVersion string) queue.SubmitCommand {
return queue.SubmitCommand{
Envelope: queue.Envelope{
SchemaVersion: queue.SchemaVersion,
MessageType: queue.MessageTypeSubmitCommand,
TraceID: "trace-1",
MessageID: "message-1",
ChannelID: "channel-1",
CreatedAt: time.Now(),
},
SubmitID: "submit-1",
PhoneNumber: "18821203795",
Content: "【安徽航天信息】您的验证码是070926，有效时间30分钟。",
CMPP: queue.CMPP{
ServiceID: "SMS",
SrcID: "1069999999",
RegisteredDelivery: 1,
MsgFmt: 8,
},
Upstream: queue.UpstreamConfig{
Account: "ljcs02",
CMPPVersion: cmppVersion,
},
}
}