Merge branch 'v2-feature-main'

dobyte · dobyte · commit f9caf382108e · 2026-04-23T16:42:21.000+08:00
diff --git a/cache/memcache/go.mod b/cache/memcache/go.mod
@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	golang.org/x/sync v0.20.0
 )
 
diff --git a/cache/redis/go.mod b/cache/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/cache/redis/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/redis/go-redis/v9 v9.17.2
 	golang.org/x/sync v0.20.0
 )
diff --git a/cluster/gate/provider.go b/cluster/gate/provider.go
@@ -92,7 +92,7 @@ func (p *provider) Broadcast(ctx context.Context, kind session.Kind, disconnect
 }
 
 // Publish 发布频道消息
-func (p *provider) Publish(ctx context.Context, channel string, disconnect bool, message []byte) int64 {
+func (p *provider) Publish(ctx context.Context, channel string, disconnect bool, message []byte) (int64, error) {
 	return p.gate.session.Publish(channel, disconnect, message)
 }
 
diff --git a/component/http/go.mod b/component/http/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/component/http/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/go-openapi/runtime v0.28.0
 	github.com/gofiber/fiber/v3 v3.0.0-rc.3
 )
diff --git a/config/consul/go.mod b/config/consul/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/config/consul/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/hashicorp/consul/api v1.32.0
 )
 
diff --git a/config/etcd/go.mod b/config/etcd/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/config/etcd/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	go.etcd.io/etcd/api/v3 v3.6.7
 	go.etcd.io/etcd/client/v3 v3.6.7
 )
diff --git a/config/nacos/go.mod b/config/nacos/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/config/nacos/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/nacos-group/nacos-sdk-go/v2 v2.3.5
 )
 
diff --git a/core/info/info.go b/core/info/info.go
@@ -27,7 +27,7 @@ const (
 	leftBottomBorder  = "└"
 	rightBottomBorder = "┘"
 	website           = "https://github.com/dobyte/due"
-	version           = "v2.5.4"
+	version           = "v2.5.5"
 	global            = "Global"
 )
 
diff --git a/crypto/ecc/go.mod b/crypto/ecc/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/crypto/ecc/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/ethereum/go-ethereum v1.12.0
 )
 
diff --git a/crypto/rsa/go.mod b/crypto/rsa/go.mod
@@ -2,7 +2,7 @@ module github.com/dobyte/due/crypto/rsa/v2
 
 go 1.25.0
 
-require github.com/dobyte/due/v2 v2.5.4
+require github.com/dobyte/due/v2 v2.5.5
 
 require (
 	dario.cat/mergo v1.0.2 // indirect
diff --git a/eventbus/kafka/go.mod b/eventbus/kafka/go.mod
@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/IBM/sarama v1.43.3
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 )
 
 require (
diff --git a/eventbus/nats/go.mod b/eventbus/nats/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/eventbus/nats/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/nats-io/nats.go v1.37.0
 )
 
diff --git a/eventbus/redis/go.mod b/eventbus/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/eventbus/redis/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/redis/go-redis/v9 v9.17.2
 )
 
diff --git a/internal/transporter/gate/client.go b/internal/transporter/gate/client.go
@@ -224,12 +224,12 @@ func (c *Client) Publish(ctx context.Context, channel string, disconnect bool, m
 		}
 		defer res.Release()
 
-		total, err := protocol.DecodePublishRes(res.Bytes())
+		code, total, err := protocol.DecodePublishRes(res.Bytes())
 		if err != nil {
 			return 0, err
 		}
 
-		return int64(total), nil
+		return int64(total), codes.CodeToError(code)
 	} else {
 		return 0, c.cli.Send(ctx, protocol.EncodePublishReq(0, channel, disconnect, message))
 	}
diff --git a/internal/transporter/gate/provider.go b/internal/transporter/gate/provider.go
@@ -27,7 +27,7 @@ type Provider interface {
 	// Broadcast 推送广播消息
 	Broadcast(ctx context.Context, kind session.Kind, disconnect bool, message []byte) (total int64, err error)
 	// Publish 发布频道消息
-	Publish(ctx context.Context, channel string, disconnect bool, message []byte) (total int64)
+	Publish(ctx context.Context, channel string, disconnect bool, message []byte) (total int64, err error)
 	// Subscribe 订阅频道
 	Subscribe(ctx context.Context, kind session.Kind, targets []int64, channel string) error
 	// Unsubscribe 取消订阅频道
diff --git a/internal/transporter/gate/server.go b/internal/transporter/gate/server.go
@@ -196,12 +196,12 @@ func (s *Server) publish(conn *server.Conn, data []byte) error {
 		return err
 	}
 
-	total := s.provider.Publish(context.Background(), channel, disconnect, message)
+	total, err := s.provider.Publish(context.Background(), channel, disconnect, message)
 
 	if seq == 0 {
-		return nil
+		return err
 	} else {
-		return conn.Send(protocol.EncodePublishRes(seq, uint64(total)))
+		return conn.Send(protocol.EncodePublishRes(seq, codes.ErrorToCode(err), uint64(total)))
 	}
 }
 
diff --git a/internal/transporter/gate/server_test.go b/internal/transporter/gate/server_test.go
@@ -68,8 +68,8 @@ func (p *provider) Broadcast(ctx context.Context, kind session.Kind, disconnect
 }
 
 // 发布频道消息（异步）
-func (p *provider) Publish(ctx context.Context, channel string, disconnect bool, message []byte) int64 {
-	return 0
+func (p *provider) Publish(ctx context.Context, channel string, disconnect bool, message []byte) (total int64, err error) {
+	return
 }
 
 // Subscribe 订阅频道
diff --git a/internal/transporter/internal/protocol/publish.go b/internal/transporter/internal/protocol/publish.go
@@ -5,12 +5,14 @@ import (
 	"io"
 
 	"github.com/dobyte/due/v2/core/buffer"
+	"github.com/dobyte/due/v2/errors"
+	"github.com/dobyte/due/v2/internal/transporter/internal/codes"
 	"github.com/dobyte/due/v2/internal/transporter/internal/route"
 )
 
 const (
 	publishReqBytes = defaultSizeBytes + defaultHeaderBytes + defaultRouteBytes + defaultSeqBytes + b8
-	publishResBytes = defaultSizeBytes + defaultHeaderBytes + defaultRouteBytes + defaultSeqBytes + b64
+	publishResBytes = defaultSizeBytes + defaultHeaderBytes + defaultRouteBytes + defaultSeqBytes + defaultCodeBytes + b64
 )
 
 // EncodePublishReq 编码发布频道消息请求
@@ -75,28 +77,43 @@ func DecodePublishReq(data []byte) (seq uint64, channel string, disconnect bool,
 }
 
 // EncodePublishRes 编码发布频道消息响应
-// 协议：size + header + route + seq + total
-func EncodePublishRes(seq uint64, total uint64) *buffer.NocopyBuffer {
+// 协议：size + header + route + seq + code + [total]
+func EncodePublishRes(seq uint64, code uint16, total ...uint64) *buffer.NocopyBuffer {
 	writer := buffer.MallocWriter(publishResBytes)
 	writer.WriteUint32s(binary.BigEndian, uint32(publishResBytes-defaultSizeBytes))
 	writer.WriteUint8s(dataBit)
 	writer.WriteUint8s(route.Publish)
 	writer.WriteUint64s(binary.BigEndian, seq)
-	writer.WriteUint64s(binary.BigEndian, total)
+	writer.WriteUint16s(binary.BigEndian, code)
+
+	if code == codes.OK && len(total) > 0 && total[0] != 0 {
+		writer.WriteUint64s(binary.BigEndian, total[0])
+	}
 
 	return buffer.NewNocopyBuffer(writer)
 }
 
 // DecodeMulticastRes 解码组播响应
 // 协议：size + header + route + seq + code + [total]
-func DecodePublishRes(data []byte) (total uint64, err error) {
+func DecodePublishRes(data []byte) (code uint16, total uint64, err error) {
+	if len(data) != publishResBytes && len(data) != publishResBytes-b64 {
+		err = errors.ErrInvalidMessage
+		return
+	}
+
 	reader := buffer.NewReader(data)
 
 	if _, err = reader.Seek(defaultSizeBytes+defaultHeaderBytes+defaultRouteBytes+defaultSeqBytes, io.SeekStart); err != nil {
 		return
 	}
 
-	total, err = reader.ReadUint64(binary.BigEndian)
+	if code, err = reader.ReadUint16(binary.BigEndian); err != nil {
+		return
+	}
+
+	if code == codes.OK && len(data) == publishResBytes {
+		total, err = reader.ReadUint64(binary.BigEndian)
+	}
 
 	return
 }
diff --git a/internal/transporter/internal/protocol/publish_test.go b/internal/transporter/internal/protocol/publish_test.go
@@ -47,18 +47,19 @@ func TestDecodePublishReq(t *testing.T) {
 }
 
 func TestEncodePublishRes(t *testing.T) {
-	buf := protocol.EncodePublishRes(1, 10)
+	buf := protocol.EncodePublishRes(1, 0, 1)
 
 	t.Log(buf.Bytes())
 }
 
 func TestDecodePublishRes(t *testing.T) {
-	buf := protocol.EncodePublishRes(1, 10)
+	buf := protocol.EncodePublishRes(1, 1)
 
-	total, err := protocol.DecodePublishRes(buf.Bytes())
+	code, total, err := protocol.DecodePublishRes(buf.Bytes())
 	if err != nil {
 		t.Fatal(err)
 	}
 
+	t.Logf("code: %v", code)
 	t.Logf("total: %v", total)
 }
diff --git a/locate/redis/go.mod b/locate/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/locate/redis/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/redis/go-redis/v9 v9.17.2
 	golang.org/x/sync v0.20.0
 )
diff --git a/lock/memcache/go.mod b/lock/memcache/go.mod
@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 )
 
 require (
diff --git a/lock/redis/go.mod b/lock/redis/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/lock/redis/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/redis/go-redis/v9 v9.17.2
 )
 
diff --git a/log/aliyun/go.mod b/log/aliyun/go.mod
@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/aliyun/aliyun-log-go-sdk v0.1.114
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 )
 
 require (
diff --git a/log/tencent/go.mod b/log/tencent/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/log/tencent/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/tencentcloud/tencentcloud-cls-sdk-go v1.0.14
 )
 
diff --git a/network/kcp/go.mod b/network/kcp/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/network/kcp/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/xtaci/kcp-go/v5 v5.6.20
 )
 
diff --git a/network/tcp/go.mod b/network/tcp/go.mod
@@ -2,7 +2,7 @@ module github.com/dobyte/due/network/tcp/v2
 
 go 1.25.0
 
-require github.com/dobyte/due/v2 v2.5.4
+require github.com/dobyte/due/v2 v2.5.5
 
 require (
 	dario.cat/mergo v1.0.2 // indirect
diff --git a/network/tcp/server_options.go b/network/tcp/server_options.go
@@ -128,7 +128,7 @@ func WithServerCredentials(certFile, keyFile string) ServerOption {
 func WithServerMaxConnNum(maxConnNum int) ServerOption {
 	return func(o *serverOptions) {
 		if maxConnNum > 0 {
-			o.writeQueueSize = maxConnNum
+			o.maxConnNum = maxConnNum
 		} else {
 			log.Warnf("the specified maxConnNum is less than zero and will be ignored")
 		}
diff --git a/network/ws/go.mod b/network/ws/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/network/ws/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/gorilla/websocket v1.5.3
 )
 
diff --git a/network/ws/server_options.go b/network/ws/server_options.go
@@ -164,7 +164,7 @@ func WithServerCheckOrigin(checkOrigin CheckOriginFunc) ServerOption {
 func WithServerMaxConnNum(maxConnNum int) ServerOption {
 	return func(o *serverOptions) {
 		if maxConnNum > 0 {
-			o.writeQueueSize = maxConnNum
+			o.maxConnNum = maxConnNum
 		} else {
 			log.Warnf("the specified maxConnNum is less than zero and will be ignored")
 		}
diff --git a/registry/consul/go.mod b/registry/consul/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/consul/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/hashicorp/consul/api v1.32.0
 )
 
diff --git a/registry/etcd/go.mod b/registry/etcd/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/etcd/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	go.etcd.io/etcd/api/v3 v3.6.7
 	go.etcd.io/etcd/client/v3 v3.6.7
 )
diff --git a/registry/nacos/go.mod b/registry/nacos/go.mod
@@ -3,7 +3,7 @@ module github.com/dobyte/due/registry/nacos/v2
 go 1.25.0
 
 require (
-	github.com/dobyte/due/v2 v2.5.4
+	github.com/dobyte/due/v2 v2.5.5
 	github.com/nacos-group/nacos-sdk-go/v2 v2.3.5
 	golang.org/x/sync v0.20.0
 )
diff --git a/session/session.go b/session/session.go
diff --git a/transport/grpc/go.mod b/transport/grpc/go.mod
diff --git a/transport/rpcx/go.mod b/transport/rpcx/go.mod

Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ go 1.25.0`
`4`	`4`
`5`	`5`	`require (`
`6`	`6`	`github.com/bradfitz/gomemcache v0.0.0-20250403215159-8d39553ac7cf`
`7`		`- github.com/dobyte/due/v2 v2.5.4`
	`7`	`+ github.com/dobyte/due/v2 v2.5.5`
`8`	`8`	`golang.org/x/sync v0.20.0`
`9`	`9`	`)`
`10`	`10`
Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@ module github.com/dobyte/due/cache/redis/v2`
`3`	`3`	`go 1.25.0`
`4`	`4`
`5`	`5`	`require (`
`6`		`- github.com/dobyte/due/v2 v2.5.4`
	`6`	`+ github.com/dobyte/due/v2 v2.5.5`
`7`	`7`	`github.com/redis/go-redis/v9 v9.17.2`
`8`	`8`	`golang.org/x/sync v0.20.0`
`9`	`9`	`)`
Original file line number	Diff line number	Diff line change
`@@ -92,7 +92,7 @@ func (p *provider) Broadcast(ctx context.Context, kind session.Kind, disconnect`
`92`	`92`	`}`
`93`	`93`
`94`	`94`	`// Publish 发布频道消息`
`95`		`-func (p *provider) Publish(ctx context.Context, channel string, disconnect bool, message []byte) int64 {`
	`95`	`+func (p *provider) Publish(ctx context.Context, channel string, disconnect bool, message []byte) (int64, error) {`
`96`	`96`	`return p.gate.session.Publish(channel, disconnect, message)`
`97`	`97`	`}`
`98`	`98`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ const (`
`27`	`27`	`leftBottomBorder = "└"`
`28`	`28`	`rightBottomBorder = "┘"`
`29`	`29`	`website = "https://github.com/dobyte/due"`
`30`		`- version = "v2.5.4"`
	`30`	`+ version = "v2.5.5"`
`31`	`31`	`global = "Global"`
`32`	`32`	`)`
`33`	`33`