add: transport ws

pkg/utils/dir_test.go

@@ -1,12 +1,13 @@
 package utils
 
 import (
-	"bou.ke/monkey"
 	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
 
+	"bou.ke/monkey"
+
 	"github.com/stretchr/testify/assert"
 )
 

plugins/transport/websocket/client.go

@@ -6,11 +6,8 @@ import (
 	"net/url"
 	"time"
 
-	"github.com/go-kratos/kratos/v2/encoding"
-
+	"github.com/apus-run/sea-kit/encoding"
 	ws "github.com/gorilla/websocket"
-
-	"github.com/tx7do/kratos-transport/broker"
 )
 
 type ClientMessageHandler func(MessagePayload) error
@@ -123,7 +120,7 @@ func (c *Client) marshalMessage(messageType MessageType, message MessagePayload)
 	case PayloadTypeBinary:
 		var msg BinaryMessage
 		msg.Type = messageType
-		msg.Body, err = broker.Marshal(c.codec, message)
+		msg.Body, err = Marshal(c.codec, message)
 		if err != nil {
 			return nil, err
 		}
@@ -137,7 +134,7 @@ func (c *Client) marshalMessage(messageType MessageType, message MessagePayload)
 		var buf []byte
 		var msg TextMessage
 		msg.Type = messageType
-		buf, err = broker.Marshal(c.codec, message)
+		buf, err = Marshal(c.codec, message)
 		msg.Body = string(buf)
 		if err != nil {
 			return nil, err
@@ -257,7 +254,7 @@ func (c *Client) unmarshalMessage(buf []byte) (*ClientHandlerData, MessagePayloa
 			payload = msg.Body
 		}
 
-		if err := broker.Unmarshal(c.codec, msg.Body, &payload); err != nil {
+		if err := Unmarshal(c.codec, msg.Body, &payload); err != nil {
 			LogErrorf("unmarshal message exception: %s", err)
 			return nil, nil, err
 		}
@@ -283,7 +280,7 @@ func (c *Client) unmarshalMessage(buf []byte) (*ClientHandlerData, MessagePayloa
 			payload = msg.Body
 		}
 
-		if err := broker.Unmarshal(c.codec, []byte(msg.Body), &payload); err != nil {
+		if err := Unmarshal(c.codec, []byte(msg.Body), &payload); err != nil {
 			LogErrorf("unmarshal message exception: %s", err)
 			return nil, nil, err
 		}

plugins/transport/websocket/encoding.go

@@ -1 +1,50 @@
 package websocket
+
+import (
+	"bytes"
+	"encoding/gob"
+	"errors"
+
+	"github.com/apus-run/sea-kit/encoding"
+	_ "github.com/apus-run/sea-kit/encoding/json"
+	_ "github.com/apus-run/sea-kit/encoding/proto"
+)
+
+func Marshal(codec encoding.Codec, msg Any) ([]byte, error) {
+	if msg == nil {
+		return nil, errors.New("message is nil")
+	}
+
+	if codec != nil {
+		dataBuffer, err := codec.Marshal(msg)
+		if err != nil {
+			return nil, err
+		}
+		return dataBuffer, nil
+	} else {
+		switch t := msg.(type) {
+		case []byte:
+			return t, nil
+		case string:
+			return []byte(t), nil
+		default:
+			var buf bytes.Buffer
+			enc := gob.NewEncoder(&buf)
+			if err := enc.Encode(msg); err != nil {
+				return nil, err
+			}
+			return buf.Bytes(), nil
+		}
+	}
+}
+
+func Unmarshal(codec encoding.Codec, inputData []byte, outValue interface{}) error {
+	if codec != nil {
+		if err := codec.Unmarshal(inputData, outValue); err != nil {
+			return err
+		}
+	} else if outValue == nil {
+		outValue = inputData
+	}
+	return nil
+}

plugins/transport/websocket/go.mod

@@ -1,42 +1,24 @@
-module github.com/tx7do/kratos-transport/transport/websocket
+module github.com/apus-run/gaia/plugins/transport/websocket
 
 go 1.19
 
 require (
-	github.com/go-kratos/kratos/v2 v2.7.0
+	github.com/apus-run/gaia v1.9.0
+	github.com/apus-run/sea-kit/encoding v0.0.0-20230930061415-4e76dbc0e8a9
+	github.com/apus-run/sea-kit/log v0.0.0-20230930061415-4e76dbc0e8a9
 	github.com/google/uuid v1.3.1
 	github.com/gorilla/websocket v1.5.0
 	github.com/stretchr/testify v1.8.4
-	github.com/tx7do/kratos-transport v1.0.12
 )
 
 require (
-	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/go-logr/logr v1.2.4 // indirect
-	github.com/go-logr/stdr v1.2.2 // indirect
-	github.com/go-playground/form/v4 v4.2.1 // indirect
-	github.com/golang/protobuf v1.5.3 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.17.1 // indirect
-	github.com/openzipkin/zipkin-go v0.4.2 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	go.opentelemetry.io/otel v1.17.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.17.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.17.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.17.0 // indirect
-	go.opentelemetry.io/otel/exporters/zipkin v1.17.0 // indirect
-	go.opentelemetry.io/otel/metric v1.17.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.17.0 // indirect
-	go.opentelemetry.io/otel/trace v1.17.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
-	golang.org/x/net v0.15.0 // indirect
-	golang.org/x/sys v0.12.0 // indirect
-	golang.org/x/text v0.13.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
-	google.golang.org/grpc v1.58.0 // indirect
+	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
-replace github.com/tx7do/kratos-transport => ../../
+replace github.com/apus-run/gaia => ../../../

plugins/transport/websocket/go.sum

@@ -0,0 +1,39 @@
+github.com/apus-run/sea-kit/encoding v0.0.0-20230930061415-4e76dbc0e8a9 h1:rsDikhV1ckW+GxFi5Vw9wG10rInlnS45TKK29oFF/g8=
+github.com/apus-run/sea-kit/encoding v0.0.0-20230930061415-4e76dbc0e8a9/go.mod h1:IUsjPu4FzcbbHR2exjACyTa4XCxQr8b9JdVkzM4r3xQ=
+github.com/apus-run/sea-kit/log v0.0.0-20230930061415-4e76dbc0e8a9 h1:fQLt1+Z1ifMgvONnkT7kFBQZcUttc8YqYBr+SjrqUXE=
+github.com/apus-run/sea-kit/log v0.0.0-20230930061415-4e76dbc0e8a9/go.mod h1:bkjkCOCQbbVy8HJbZ8HpVZ8yR36L9esmhEu869idCc8=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
+github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
+google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

plugins/transport/websocket/logger.go

@@ -3,7 +3,7 @@ package websocket
 import (
 	"fmt"
 
-	"github.com/go-kratos/kratos/v2/log"
+	"github.com/apus-run/sea-kit/log"
 )
 
 const (

plugins/transport/websocket/options.go

@@ -3,9 +3,11 @@ package websocket
 import (
 	"crypto/tls"
 	"net"
+	"net/http"
 	"time"
 
-	"github.com/go-kratos/kratos/v2/encoding"
+	"github.com/apus-run/sea-kit/encoding"
+	ws "github.com/gorilla/websocket"
 )
 
 type PayloadType uint8
@@ -15,6 +17,58 @@ const (
 	PayloadTypeText   = 1
 )
 
+type Server struct {
+	*http.Server
+
+	lis      net.Listener
+	tlsConf  *tls.Config
+	upgrader *ws.Upgrader
+
+	network     string
+	address     string
+	path        string
+	strictSlash bool
+
+	timeout time.Duration
+
+	err   error
+	codec encoding.Codec
+
+	messageHandlers MessageHandlerMap
+
+	sessionMgr *SessionManager
+
+	register   chan *Session
+	unregister chan *Session
+
+	payloadType PayloadType
+}
+
+// defaultServer return a default config server
+func defaultServer() *Server {
+	return &Server{
+		network:     "tcp",
+		address:     ":0",
+		timeout:     1 * time.Second,
+		strictSlash: true,
+		path:        "/",
+
+		messageHandlers: make(MessageHandlerMap),
+
+		sessionMgr: NewSessionManager(),
+		upgrader: &ws.Upgrader{
+			ReadBufferSize:  1024,
+			WriteBufferSize: 1024,
+			CheckOrigin:     func(r *http.Request) bool { return true },
+		},
+
+		register:   make(chan *Session),
+		unregister: make(chan *Session),
+
+		payloadType: PayloadTypeBinary,
+	}
+}
+
 type ServerOption func(o *Server)
 
 func WithNetwork(network string) ServerOption {