更新依赖和调整目录结构,未修改模块
This commit is contained in:
130
pkgs/ecmall/internal/config/config.go
Normal file
130
pkgs/ecmall/internal/config/config.go
Normal file
@@ -0,0 +1,130 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"net"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
ftsService "bsm/full/module/base/fts/service"
|
||||
mgtService "bsm/full/module/base/mgt/service"
|
||||
passportService "bsm/full/module/base/passport/service"
|
||||
senderService "bsm/full/module/base/sender/service"
|
||||
walletService "bsm/full/module/finance/wallet/service"
|
||||
"git.apinb.com/bsm-sdk/core/conf"
|
||||
"git.apinb.com/bsm-sdk/core/env"
|
||||
"git.apinb.com/bsm-sdk/core/printer"
|
||||
coreVars "git.apinb.com/bsm-sdk/core/vars"
|
||||
)
|
||||
|
||||
// SrvConfig is the ecmall aggregate configuration. It only carries the
|
||||
// services registered by pkgs/ecmall: the e-commerce domain (address, mall,
|
||||
// market, order), the payment domain (wallet), the platform domain (ads, cms,
|
||||
// feedback, fts, logs, mgt) and the base support services they rely on
|
||||
// (initial, passport, sender). Modules that are not part of ecmall keep their
|
||||
// own configuration in their own process.
|
||||
type SrvConfig struct {
|
||||
conf.Base `yaml:",inline"`
|
||||
Server ServerConfig `yaml:"Server"`
|
||||
Authorization AuthorizationConfig `yaml:"Authorization"`
|
||||
Databases *conf.DBConf `yaml:"Databases"`
|
||||
MicroService *conf.MicroServiceConf `yaml:"MicroService"`
|
||||
Rpc map[string]conf.RpcConf `yaml:"Rpc"`
|
||||
Apm *conf.ApmConf `yaml:"APM"`
|
||||
Etcd *conf.EtcdConf `yaml:"Etcd"`
|
||||
Services []string `yaml:"Services"`
|
||||
Fts *ftsService.Config `yaml:"Fts"`
|
||||
Mgt *mgtService.Config `yaml:"Mgt"`
|
||||
Passport *passportService.Config `yaml:"Passport"`
|
||||
Sender *senderService.Config `yaml:"Sender"`
|
||||
Wallet *walletService.Config `yaml:"Wallet"`
|
||||
}
|
||||
|
||||
type ListenerConfig struct {
|
||||
BindIP string `yaml:"BindIP"`
|
||||
Port string `yaml:"Port"`
|
||||
Addr string `yaml:"-"`
|
||||
}
|
||||
|
||||
type ServerConfig struct {
|
||||
GRPC ListenerConfig `yaml:"GRPC"`
|
||||
HTTP ListenerConfig `yaml:"HTTP"`
|
||||
}
|
||||
|
||||
type AuthorizationConfig struct {
|
||||
Key string `yaml:"Key"`
|
||||
Expire int64 `yaml:"Expire"`
|
||||
Anonymous []string `yaml:"Anonymous"`
|
||||
}
|
||||
|
||||
var Spec SrvConfig
|
||||
|
||||
func New(serviceKey string) {
|
||||
conf.New(serviceKey, &Spec)
|
||||
normalizeListener(&Spec.Server.GRPC)
|
||||
normalizeListener(&Spec.Server.HTTP)
|
||||
if Spec.Server.GRPC.Addr == Spec.Server.HTTP.Addr {
|
||||
panic("gRPC and HTTP listeners must use different addresses")
|
||||
}
|
||||
if strings.TrimSpace(Spec.Authorization.Key) == "" {
|
||||
panic("Authorization.Key must not be empty")
|
||||
}
|
||||
keyLength := len(Spec.Authorization.Key)
|
||||
if keyLength != 16 && keyLength != 24 && keyLength != 32 {
|
||||
panic("Authorization.Key must contain 16, 24, or 32 bytes")
|
||||
}
|
||||
if Spec.Authorization.Expire <= 0 {
|
||||
panic("Authorization.Expire must be greater than zero")
|
||||
}
|
||||
env.NewEnv().JwtSecretKey = Spec.Authorization.Key
|
||||
coreVars.JwtExpire = time.Duration(Spec.Authorization.Expire) * time.Second
|
||||
// Keep the embedded base address meaningful for module configurations that
|
||||
// still consume it, while ecmall itself uses the two explicit listeners.
|
||||
Spec.BindIP, Spec.Port, Spec.Addr = Spec.Server.HTTP.BindIP, Spec.Server.HTTP.Port, Spec.Server.HTTP.Addr
|
||||
conf.NotNil(Spec.Service, Spec.Cache)
|
||||
assignSharedConfig()
|
||||
conf.PrintInfo(Spec.Server.GRPC.Addr)
|
||||
printer.Info("gRPC Address: %s", Spec.Server.GRPC.Addr)
|
||||
printer.Info("HTTP Address: %s", Spec.Server.HTTP.Addr)
|
||||
}
|
||||
|
||||
func normalizeListener(listener *ListenerConfig) {
|
||||
listener.Port = conf.CheckPort(listener.Port)
|
||||
listener.BindIP = conf.CheckIP(listener.BindIP)
|
||||
listener.Addr = net.JoinHostPort(listener.BindIP, listener.Port)
|
||||
}
|
||||
|
||||
func assignSharedConfig() {
|
||||
if Spec.Fts != nil {
|
||||
Spec.Fts.Base, Spec.Fts.Databases, Spec.Fts.Rpc, Spec.Fts.Apm, Spec.Fts.Etcd = Spec.Base, Spec.Databases, Spec.Rpc, Spec.Apm, Spec.Etcd
|
||||
}
|
||||
if Spec.Mgt != nil {
|
||||
Spec.Mgt.Base, Spec.Mgt.Databases, Spec.Mgt.Rpc, Spec.Mgt.Apm, Spec.Mgt.Etcd = Spec.Base, Spec.Databases, Spec.Rpc, Spec.Apm, Spec.Etcd
|
||||
}
|
||||
if Spec.Passport != nil {
|
||||
Spec.Passport.Base, Spec.Passport.Databases, Spec.Passport.MicroService, Spec.Passport.Rpc, Spec.Passport.Apm, Spec.Passport.Etcd = Spec.Base, Spec.Databases, Spec.MicroService, Spec.Rpc, Spec.Apm, Spec.Etcd
|
||||
}
|
||||
if Spec.Sender != nil {
|
||||
Spec.Sender.Base, Spec.Sender.Databases, Spec.Sender.MicroService, Spec.Sender.Rpc, Spec.Sender.Apm, Spec.Sender.Etcd = Spec.Base, Spec.Databases, Spec.MicroService, Spec.Rpc, Spec.Apm, Spec.Etcd
|
||||
}
|
||||
if Spec.Wallet != nil {
|
||||
Spec.Wallet.Base, Spec.Wallet.Databases, Spec.Wallet.MicroService, Spec.Wallet.Rpc, Spec.Wallet.Apm, Spec.Wallet.Etcd = Spec.Base, Spec.Databases, Spec.MicroService, Spec.Rpc, Spec.Apm, Spec.Etcd
|
||||
}
|
||||
}
|
||||
|
||||
func Enabled(name string) bool {
|
||||
if raw := strings.TrimSpace(os.Getenv("BSM_SERVICES")); raw != "" {
|
||||
for _, selected := range strings.Split(raw, ",") {
|
||||
if strings.EqualFold(strings.TrimSpace(selected), name) || strings.EqualFold(strings.TrimSpace(selected), "all") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
for _, service := range Spec.Services {
|
||||
if strings.EqualFold(strings.TrimSpace(service), name) || strings.EqualFold(strings.TrimSpace(service), "all") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
53
pkgs/ecmall/internal/config/config_test.go
Normal file
53
pkgs/ecmall/internal/config/config_test.go
Normal file
@@ -0,0 +1,53 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"go.yaml.in/yaml/v3"
|
||||
)
|
||||
|
||||
// ecmallServices is the aggregate registry of pkgs/ecmall. Keep it in sync with
|
||||
// the services table in internal/service.
|
||||
var ecmallServices = []string{"address", "ads", "cms", "feedback", "fts", "initial", "logs", "mgt", "mall", "market", "order", "passport", "sender", "wallet"}
|
||||
|
||||
func TestEcmallDevConfig(t *testing.T) {
|
||||
data, err := os.ReadFile(filepath.Join("..", "..", "etc", "default_dev.yaml"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var cfg SrvConfig
|
||||
if err := yaml.Unmarshal(data, &cfg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(cfg.Services) == 0 {
|
||||
t.Fatal("services must not be empty")
|
||||
}
|
||||
if cfg.Server.GRPC.Port == "" || cfg.Server.HTTP.Port == "" || cfg.Server.GRPC.Port == cfg.Server.HTTP.Port {
|
||||
t.Fatal("separate gRPC and HTTP ports are required")
|
||||
}
|
||||
if cfg.Authorization.Key == "" || cfg.Authorization.Expire <= 0 {
|
||||
t.Fatal("authorization key and expiration are required")
|
||||
}
|
||||
keyLength := len(cfg.Authorization.Key)
|
||||
if keyLength != 16 && keyLength != 24 && keyLength != 32 {
|
||||
t.Fatal("authorization key must be compatible with the JWT issuer")
|
||||
}
|
||||
if cfg.Fts == nil || cfg.Mgt == nil || cfg.Passport == nil || cfg.Sender == nil || cfg.Wallet == nil {
|
||||
t.Fatal("service-specific configuration is incomplete")
|
||||
}
|
||||
|
||||
enabled := make([]string, 0, len(cfg.Services))
|
||||
for _, service := range cfg.Services {
|
||||
enabled = append(enabled, strings.ToLower(strings.TrimSpace(service)))
|
||||
}
|
||||
sort.Strings(enabled)
|
||||
expected := append([]string(nil), ecmallServices...)
|
||||
sort.Strings(expected)
|
||||
if strings.Join(enabled, ",") != strings.Join(expected, ",") {
|
||||
t.Fatalf("ecmall must only enable %v, got %v", expected, enabled)
|
||||
}
|
||||
}
|
||||
42
pkgs/ecmall/internal/impl/impl.go
Normal file
42
pkgs/ecmall/internal/impl/impl.go
Normal file
@@ -0,0 +1,42 @@
|
||||
package impl
|
||||
|
||||
import (
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/cache/redis"
|
||||
"git.apinb.com/bsm-sdk/core/with"
|
||||
"github.com/patrickmn/go-cache"
|
||||
clientv3 "go.etcd.io/etcd/client/v3"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
var (
|
||||
// RedisService Redis缓存服务客户端,用于存储临时数据和缓存热点数据
|
||||
RedisService *redis.RedisClient
|
||||
|
||||
// EtcdService Etcd客户端,用于服务发现和配置管理
|
||||
EtcdService *clientv3.Client
|
||||
|
||||
// DBService 数据库服务连接, 用于数据持久化存储
|
||||
DBService *gorm.DB
|
||||
|
||||
// MemoryService 内存缓存服务, 用于存储频繁访问的数据,提供最快的访问速度
|
||||
MemoryService *cache.Cache
|
||||
)
|
||||
|
||||
// NewImpl 初始化实现层
|
||||
// 建立与各种外部服务的连接,包括内存缓存、Redis、数据库和Etcd
|
||||
// 这些连接将在整个服务生命周期中使用
|
||||
func NewImpl() {
|
||||
// 初始化内存缓存服务,用于存储热点数据
|
||||
MemoryService = with.Memory(nil)
|
||||
|
||||
// 初始化Redis缓存服务,用于分布式缓存
|
||||
RedisService = with.RedisCache(config.Spec.Cache)
|
||||
|
||||
// 初始化数据库连接,用于数据持久化
|
||||
DBService = with.Databases(config.Spec.Databases, nil)
|
||||
|
||||
// 初始化Etcd客户端,用于服务发现和配置管理
|
||||
EtcdService = with.Etcd(config.Spec.Etcd)
|
||||
}
|
||||
126
pkgs/ecmall/internal/server/authorization.go
Normal file
126
pkgs/ecmall/internal/server/authorization.go
Normal file
@@ -0,0 +1,126 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/errcode"
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/metadata"
|
||||
)
|
||||
|
||||
type authorization struct {
|
||||
key []byte
|
||||
expire time.Duration
|
||||
anonymous map[string]struct{}
|
||||
}
|
||||
|
||||
func newAuthorization(key string, expireSeconds int64, anonymous []string) (*authorization, error) {
|
||||
key = strings.TrimSpace(key)
|
||||
if key == "" {
|
||||
return nil, errcode.ErrTokenSecretKeyNotFound
|
||||
}
|
||||
if expireSeconds <= 0 {
|
||||
return nil, errcode.ErrTokenAuthExpire
|
||||
}
|
||||
allowed := make(map[string]struct{}, len(anonymous))
|
||||
for _, item := range anonymous {
|
||||
if item = normalizePath(item); item != "" {
|
||||
allowed[item] = struct{}{}
|
||||
}
|
||||
}
|
||||
return &authorization{key: []byte(key), expire: time.Duration(expireSeconds) * time.Second, anonymous: allowed}, nil
|
||||
}
|
||||
|
||||
func (a *authorization) unaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
if !a.isAnonymous(info.FullMethod) {
|
||||
values := metadata.ValueFromIncomingContext(ctx, "authorization")
|
||||
if len(values) == 0 {
|
||||
return nil, errcode.ErrHeaderAuthorization
|
||||
}
|
||||
if err := a.validate(values[0]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return handler(ctx, req)
|
||||
}
|
||||
|
||||
func (a *authorization) httpMiddleware(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
requestPath := canonicalHTTPPath(r.URL.Path)
|
||||
if !a.isAnonymous(requestPath) {
|
||||
if err := a.validate(r.Header.Get("Authorization")); err != nil {
|
||||
writeHTTPAuthorizationError(w, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
func (a *authorization) validate(raw string) error {
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" {
|
||||
return errcode.ErrHeaderAuthorization
|
||||
}
|
||||
claims := &jwt.RegisteredClaims{}
|
||||
tokenValue, err := jwt.ParseWithClaims(raw, claims, func(tokenValue *jwt.Token) (any, error) {
|
||||
if tokenValue.Method != jwt.SigningMethodHS256 {
|
||||
return nil, fmt.Errorf("unexpected signing method: %s", tokenValue.Method.Alg())
|
||||
}
|
||||
return a.key, nil
|
||||
}, jwt.WithExpirationRequired(), jwt.WithIssuedAt(), jwt.WithValidMethods([]string{jwt.SigningMethodHS256.Alg()}))
|
||||
if err != nil || !tokenValue.Valid {
|
||||
if errors.Is(err, jwt.ErrTokenExpired) {
|
||||
return errcode.ErrTokenAuthExpire
|
||||
}
|
||||
return errcode.ErrTokenAuthParseFail
|
||||
}
|
||||
if claims.IssuedAt == nil {
|
||||
return errcode.ErrTokenDataInvalid
|
||||
}
|
||||
now := time.Now()
|
||||
if claims.IssuedAt.Time.After(now) || now.Sub(claims.IssuedAt.Time) > a.expire {
|
||||
return errcode.ErrTokenAuthExpire
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *authorization) isAnonymous(requestPath string) bool {
|
||||
requestPath = normalizePath(requestPath)
|
||||
if strings.HasPrefix(requestPath, "/grpc.reflection.") || strings.HasPrefix(requestPath, "/grpc.health.") {
|
||||
return true
|
||||
}
|
||||
_, ok := a.anonymous[requestPath]
|
||||
return ok
|
||||
}
|
||||
|
||||
func canonicalHTTPPath(requestPath string) string {
|
||||
normalized := normalizePath(requestPath)
|
||||
parts := strings.Split(strings.TrimPrefix(normalized, "/"), "/")
|
||||
if len(parts) == 4 && parts[0] == "rpc" {
|
||||
return "/" + parts[1] + "." + parts[2] + "/" + parts[3]
|
||||
}
|
||||
return normalized
|
||||
}
|
||||
|
||||
func normalizePath(value string) string {
|
||||
value = strings.TrimSpace(value)
|
||||
if value == "" {
|
||||
return ""
|
||||
}
|
||||
return path.Clean("/" + strings.TrimPrefix(value, "/"))
|
||||
}
|
||||
|
||||
func writeHTTPAuthorizationError(w http.ResponseWriter, err error) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_ = json.NewEncoder(w).Encode(newErrorResponse(err))
|
||||
}
|
||||
102
pkgs/ecmall/internal/server/authorization_test.go
Normal file
102
pkgs/ecmall/internal/server/authorization_test.go
Normal file
@@ -0,0 +1,102 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/errcode"
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const testAuthorizationKey = "0123456789abcdef0123456789abcdef"
|
||||
|
||||
func TestHTTPAuthorization(t *testing.T) {
|
||||
auth, err := newAuthorization(testAuthorizationKey, 3600, []string{"/passport.Login/Pwd", "/rest/fts/ping"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
next := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })
|
||||
handler := auth.httpMiddleware(next)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
path string
|
||||
token string
|
||||
want int
|
||||
code int32
|
||||
}{
|
||||
{name: "grpc gateway anonymous", path: "/passport.Login/Pwd", want: http.StatusNoContent},
|
||||
{name: "dynamic rpc anonymous", path: "/rpc/passport/Login/Pwd", want: http.StatusNoContent},
|
||||
{name: "rest anonymous", path: "/rest/fts/ping", want: http.StatusNoContent},
|
||||
{name: "missing token", path: "/passport.Account/Get", want: http.StatusOK, code: int32(status.Code(errcode.ErrHeaderAuthorization))},
|
||||
{name: "valid raw token", path: "/passport.Account/Get", token: signedToken(t, time.Now(), time.Now().Add(time.Hour)), want: http.StatusNoContent},
|
||||
{name: "bearer rejected", path: "/passport.Account/Get", token: "Bearer " + signedToken(t, time.Now(), time.Now().Add(time.Hour)), want: http.StatusOK, code: int32(status.Code(errcode.ErrTokenAuthParseFail))},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
request := httptest.NewRequest(http.MethodPost, test.path, nil)
|
||||
if test.token != "" {
|
||||
request.Header.Set("Authorization", test.token)
|
||||
}
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, request)
|
||||
if response.Code != test.want {
|
||||
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
|
||||
}
|
||||
if test.want == http.StatusOK && !strings.Contains(response.Body.String(), `"code":`+fmt.Sprint(test.code)) {
|
||||
t.Fatalf("unexpected authorization error: %s", response.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthorizationRejectsTokenOlderThanConfiguredLifetime(t *testing.T) {
|
||||
auth, err := newAuthorization(testAuthorizationKey, 60, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := auth.validate(signedToken(t, time.Now().Add(-2*time.Minute), time.Now().Add(time.Hour))); err == nil {
|
||||
t.Fatal("expected token older than configured lifetime to be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGRPCAuthorization(t *testing.T) {
|
||||
auth, err := newAuthorization(testAuthorizationKey, 3600, []string{"/passport.Login/Pwd"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
handler := func(context.Context, any) (any, error) { return "ok", nil }
|
||||
|
||||
if _, err := auth.unaryInterceptor(context.Background(), nil, &grpc.UnaryServerInfo{FullMethod: "/passport.Login/Pwd"}, handler); err != nil {
|
||||
t.Fatalf("anonymous method failed: %v", err)
|
||||
}
|
||||
if _, err := auth.unaryInterceptor(context.Background(), nil, &grpc.UnaryServerInfo{FullMethod: "/passport.Account/Get"}, handler); status.Code(err) != status.Code(errcode.ErrHeaderAuthorization) {
|
||||
t.Fatalf("expected unauthenticated, got %v", err)
|
||||
}
|
||||
ctx := metadata.NewIncomingContext(context.Background(), metadata.Pairs("authorization", signedToken(t, time.Now(), time.Now().Add(time.Hour))))
|
||||
if _, err := auth.unaryInterceptor(ctx, nil, &grpc.UnaryServerInfo{FullMethod: "/passport.Account/Get"}, handler); err != nil {
|
||||
t.Fatalf("valid token failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func signedToken(t *testing.T, issuedAt, expiresAt time.Time) string {
|
||||
t.Helper()
|
||||
claims := jwt.RegisteredClaims{
|
||||
IssuedAt: jwt.NewNumericDate(issuedAt),
|
||||
NotBefore: jwt.NewNumericDate(issuedAt),
|
||||
ExpiresAt: jwt.NewNumericDate(expiresAt),
|
||||
}
|
||||
value, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(testAuthorizationKey))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return value
|
||||
}
|
||||
194
pkgs/ecmall/internal/server/dynamic.go
Normal file
194
pkgs/ecmall/internal/server/dynamic.go
Normal file
@@ -0,0 +1,194 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/errcode"
|
||||
"github.com/gin-gonic/gin"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/metadata"
|
||||
reflectionv1 "google.golang.org/grpc/reflection/grpc_reflection_v1"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/encoding/protojson"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"google.golang.org/protobuf/reflect/protodesc"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
"google.golang.org/protobuf/reflect/protoregistry"
|
||||
"google.golang.org/protobuf/types/descriptorpb"
|
||||
"google.golang.org/protobuf/types/dynamicpb"
|
||||
)
|
||||
|
||||
const maxDynamicRPCBody = 4 << 20
|
||||
|
||||
type dynamicGateway struct {
|
||||
conn *grpc.ClientConn
|
||||
mu sync.RWMutex
|
||||
cache map[string]protoreflect.MethodDescriptor
|
||||
}
|
||||
|
||||
type dynamicRPCResponse struct {
|
||||
Code int32 `json:"code"`
|
||||
Message string `json:"message"`
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
func newDynamicGateway(grpcAddr string) (*dynamicGateway, error) {
|
||||
conn, err := grpc.NewClient(reflectionTarget(grpcAddr), grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create dynamic gRPC client: %w", err)
|
||||
}
|
||||
return &dynamicGateway{conn: conn, cache: make(map[string]protoreflect.MethodDescriptor)}, nil
|
||||
}
|
||||
|
||||
func reflectionTarget(addr string) string {
|
||||
host, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
return addr
|
||||
}
|
||||
if host == "" || host == "0.0.0.0" || host == "::" {
|
||||
host = "127.0.0.1"
|
||||
}
|
||||
return net.JoinHostPort(host, port)
|
||||
}
|
||||
|
||||
func (g *dynamicGateway) Close() error { return g.conn.Close() }
|
||||
|
||||
func (g *dynamicGateway) handle(c *gin.Context) {
|
||||
moduleName := strings.TrimSpace(c.Param("module"))
|
||||
serviceShortName := strings.TrimSpace(c.Param("service"))
|
||||
methodName := strings.TrimSpace(c.Param("method"))
|
||||
if moduleName == "" || serviceShortName == "" || methodName == "" {
|
||||
writeDynamicError(c, errcode.String(errcode.ErrInvalidArgument, "path must be /rpc/{module}/{service}/{method}"))
|
||||
return
|
||||
}
|
||||
serviceName := moduleName + "." + serviceShortName
|
||||
descriptor, err := g.resolveMethod(c.Request.Context(), serviceName, methodName)
|
||||
if err != nil {
|
||||
writeDynamicError(c, err)
|
||||
return
|
||||
}
|
||||
if descriptor.IsStreamingClient() || descriptor.IsStreamingServer() {
|
||||
writeDynamicError(c, errcode.String(errcode.ErrUnimplemented, "streaming RPC methods are not supported"))
|
||||
return
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(c.Request.Body, maxDynamicRPCBody+1))
|
||||
if err != nil {
|
||||
writeDynamicError(c, errcode.String(errcode.ErrInvalidArgument, "read request body: "+err.Error()))
|
||||
return
|
||||
}
|
||||
if len(body) > maxDynamicRPCBody {
|
||||
writeDynamicError(c, errcode.String(errcode.ErrResourceExhausted, "request body exceeds 4 MiB"))
|
||||
return
|
||||
}
|
||||
|
||||
request := dynamicpb.NewMessage(descriptor.Input())
|
||||
if err := (protojson.UnmarshalOptions{DiscardUnknown: false}).Unmarshal(body, request); err != nil {
|
||||
writeDynamicError(c, errcode.String(errcode.ErrJsonUnmarshal, err.Error()))
|
||||
return
|
||||
}
|
||||
response := dynamicpb.NewMessage(descriptor.Output())
|
||||
ctx := outgoingMetadata(c.Request)
|
||||
grpcMethod := "/" + serviceName + "/" + methodName
|
||||
if err := g.conn.Invoke(ctx, grpcMethod, request, response); err != nil {
|
||||
writeDynamicError(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
data, err := (protojson.MarshalOptions{UseProtoNames: false}).Marshal(response)
|
||||
if err != nil {
|
||||
writeDynamicError(c, errcode.String(errcode.ErrJsonMarshal, err.Error()))
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, dynamicRPCResponse{Code: int32(codes.OK), Message: codes.OK.String(), Data: data})
|
||||
}
|
||||
|
||||
func (g *dynamicGateway) resolveMethod(ctx context.Context, serviceName, methodName string) (protoreflect.MethodDescriptor, error) {
|
||||
cacheKey := serviceName + "." + methodName
|
||||
g.mu.RLock()
|
||||
method := g.cache[cacheKey]
|
||||
g.mu.RUnlock()
|
||||
if method != nil {
|
||||
return method, nil
|
||||
}
|
||||
|
||||
stream, err := reflectionv1.NewServerReflectionClient(g.conn).ServerReflectionInfo(ctx)
|
||||
if err != nil {
|
||||
return nil, errcode.String(errcode.ErrUnavailable, "open gRPC reflection stream: "+err.Error())
|
||||
}
|
||||
if err := stream.Send(&reflectionv1.ServerReflectionRequest{
|
||||
MessageRequest: &reflectionv1.ServerReflectionRequest_FileContainingSymbol{FileContainingSymbol: serviceName},
|
||||
}); err != nil {
|
||||
return nil, errcode.String(errcode.ErrUnavailable, "query gRPC reflection: "+err.Error())
|
||||
}
|
||||
reflectionResponse, err := stream.Recv()
|
||||
if err != nil {
|
||||
return nil, errcode.String(errcode.ErrUnavailable, "read gRPC reflection response: "+err.Error())
|
||||
}
|
||||
fileResponse := reflectionResponse.GetFileDescriptorResponse()
|
||||
if fileResponse == nil {
|
||||
if reflectionErr := reflectionResponse.GetErrorResponse(); reflectionErr != nil {
|
||||
return nil, sdkError(status.Error(codes.Code(reflectionErr.ErrorCode), reflectionErr.ErrorMessage))
|
||||
}
|
||||
return nil, errcode.String(errcode.ErrRecordNotFound, "service descriptor not found")
|
||||
}
|
||||
|
||||
set := &descriptorpb.FileDescriptorSet{}
|
||||
for _, encoded := range fileResponse.FileDescriptorProto {
|
||||
file := &descriptorpb.FileDescriptorProto{}
|
||||
if err := proto.Unmarshal(encoded, file); err != nil {
|
||||
return nil, errcode.String(errcode.ErrInternal, "decode reflected descriptor: "+err.Error())
|
||||
}
|
||||
set.File = append(set.File, file)
|
||||
}
|
||||
files, err := protodesc.NewFiles(set)
|
||||
if err != nil {
|
||||
return nil, errcode.String(errcode.ErrInternal, "build reflected descriptors: "+err.Error())
|
||||
}
|
||||
descriptor, err := files.FindDescriptorByName(protoreflect.FullName(serviceName))
|
||||
if err != nil {
|
||||
if err == protoregistry.NotFound {
|
||||
return nil, errcode.String(errcode.ErrRecordNotFound, "service not found")
|
||||
}
|
||||
return nil, errcode.String(errcode.ErrInternal, "resolve service descriptor: "+err.Error())
|
||||
}
|
||||
service, ok := descriptor.(protoreflect.ServiceDescriptor)
|
||||
if !ok {
|
||||
return nil, errcode.String(errcode.ErrRecordNotFound, "symbol is not a gRPC service")
|
||||
}
|
||||
method = service.Methods().ByName(protoreflect.Name(methodName))
|
||||
if method == nil {
|
||||
return nil, errcode.String(errcode.ErrRecordNotFound, "method not found")
|
||||
}
|
||||
g.mu.Lock()
|
||||
g.cache[cacheKey] = method
|
||||
g.mu.Unlock()
|
||||
return method, nil
|
||||
}
|
||||
|
||||
func outgoingMetadata(request *http.Request) context.Context {
|
||||
pairs := make([]string, 0)
|
||||
for name, values := range request.Header {
|
||||
lower := strings.ToLower(name)
|
||||
if lower != "authorization" && lower != "x-request-id" && !strings.HasPrefix(lower, "x-") {
|
||||
continue
|
||||
}
|
||||
for _, value := range values {
|
||||
pairs = append(pairs, lower, value)
|
||||
}
|
||||
}
|
||||
return metadata.NewOutgoingContext(request.Context(), metadata.Pairs(pairs...))
|
||||
}
|
||||
|
||||
func writeDynamicError(c *gin.Context, err error) {
|
||||
c.JSON(http.StatusOK, newErrorResponse(sdkError(err)))
|
||||
}
|
||||
76
pkgs/ecmall/internal/server/dynamic_test.go
Normal file
76
pkgs/ecmall/internal/server/dynamic_test.go
Normal file
@@ -0,0 +1,76 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/health"
|
||||
healthpb "google.golang.org/grpc/health/grpc_health_v1"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/reflection"
|
||||
)
|
||||
|
||||
func TestDynamicGatewayInvokesUnaryRPC(t *testing.T) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
grpcServer := grpc.NewServer()
|
||||
healthServer := health.NewServer()
|
||||
healthServer.SetServingStatus("", healthpb.HealthCheckResponse_SERVING)
|
||||
healthpb.RegisterHealthServer(grpcServer, healthServer)
|
||||
reflection.Register(grpcServer)
|
||||
go func() { _ = grpcServer.Serve(listener) }()
|
||||
t.Cleanup(func() {
|
||||
grpcServer.Stop()
|
||||
_ = listener.Close()
|
||||
})
|
||||
|
||||
gateway, err := newDynamicGateway(listener.Addr().String())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = gateway.Close() })
|
||||
|
||||
gin.SetMode(gin.TestMode)
|
||||
engine := gin.New()
|
||||
engine.POST("/rpc/:module/:service/:method", gateway.handle)
|
||||
request := httptest.NewRequest(http.MethodPost, "/rpc/grpc.health.v1/Health/Check", strings.NewReader(`{"service":""}`))
|
||||
response := httptest.NewRecorder()
|
||||
engine.ServeHTTP(response, request)
|
||||
|
||||
if response.Code != http.StatusOK {
|
||||
t.Fatalf("unexpected HTTP status: %d", response.Code)
|
||||
}
|
||||
var payload struct {
|
||||
Code int32 `json:"code"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if payload.Code != 0 || !strings.Contains(string(payload.Data), `"SERVING"`) {
|
||||
t.Fatalf("unexpected response: %s", response.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestOutgoingMetadataFiltersHeaders(t *testing.T) {
|
||||
request := httptest.NewRequest(http.MethodPost, "/", nil)
|
||||
request.Header.Set("Authorization", "Bearer token")
|
||||
request.Header.Set("X-Request-ID", "request-id")
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
ctx := outgoingMetadata(request)
|
||||
forwarded, ok := metadata.FromOutgoingContext(ctx)
|
||||
if !ok || len(forwarded.Get("authorization")) != 1 || len(forwarded.Get("x-request-id")) != 1 {
|
||||
t.Fatalf("expected forwarded metadata: %v", forwarded)
|
||||
}
|
||||
if len(forwarded.Get("content-type")) != 0 {
|
||||
t.Fatalf("content-type must not be forwarded: %v", forwarded)
|
||||
}
|
||||
}
|
||||
78
pkgs/ecmall/internal/server/response.go
Normal file
78
pkgs/ecmall/internal/server/response.go
Normal file
@@ -0,0 +1,78 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/errcode"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
type errorResponse struct {
|
||||
Code int32 `json:"code"`
|
||||
Message string `json:"message"`
|
||||
Details any `json:"details"`
|
||||
Timeseq int64 `json:"timeseq"`
|
||||
}
|
||||
|
||||
func sdkError(err error) error {
|
||||
code := status.Code(err)
|
||||
// SDK business errors already use custom codes and must pass through.
|
||||
if code > codes.Unauthenticated {
|
||||
return err
|
||||
}
|
||||
message := status.Convert(err).Message()
|
||||
var target error
|
||||
switch code {
|
||||
case codes.Canceled:
|
||||
target = errcode.ErrCanceled
|
||||
case codes.InvalidArgument:
|
||||
target = errcode.ErrInvalidArgument
|
||||
case codes.DeadlineExceeded:
|
||||
target = errcode.ErrDeadlineExceeded
|
||||
case codes.NotFound:
|
||||
target = errcode.ErrRecordNotFound
|
||||
case codes.AlreadyExists:
|
||||
target = errcode.ErrAlreadyExists
|
||||
case codes.PermissionDenied:
|
||||
target = errcode.ErrPermissionDenied
|
||||
case codes.ResourceExhausted:
|
||||
target = errcode.ErrResourceExhausted
|
||||
case codes.FailedPrecondition:
|
||||
target = errcode.ErrFailedPrecondition
|
||||
case codes.Aborted:
|
||||
target = errcode.ErrAborted
|
||||
case codes.OutOfRange:
|
||||
target = errcode.ErrOutOfRange
|
||||
case codes.Unimplemented:
|
||||
target = errcode.ErrUnimplemented
|
||||
case codes.Unavailable:
|
||||
target = errcode.ErrUnavailable
|
||||
case codes.DataLoss:
|
||||
target = errcode.ErrDataLoss
|
||||
case codes.Unauthenticated:
|
||||
target = errcode.ErrUnauthenticated
|
||||
case codes.Internal:
|
||||
target = errcode.ErrInternal
|
||||
default:
|
||||
target = errcode.ErrUnknown
|
||||
}
|
||||
if message == "" || message == status.Convert(target).Message() {
|
||||
return target
|
||||
}
|
||||
return errcode.String(target, message)
|
||||
}
|
||||
|
||||
func newErrorResponse(err error) errorResponse {
|
||||
response := errorResponse{
|
||||
Code: 500,
|
||||
Message: err.Error(),
|
||||
Details: "",
|
||||
Timeseq: time.Now().UnixMilli(),
|
||||
}
|
||||
if grpcStatus, ok := status.FromError(err); ok {
|
||||
response.Code = int32(grpcStatus.Code())
|
||||
response.Message = grpcStatus.Message()
|
||||
}
|
||||
return response
|
||||
}
|
||||
133
pkgs/ecmall/internal/server/server.go
Normal file
133
pkgs/ecmall/internal/server/server.go
Normal file
@@ -0,0 +1,133 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
gwRuntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/reflection"
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
GRPC *grpc.Server
|
||||
Gateway *gwRuntime.ServeMux
|
||||
HTTP *gin.Engine
|
||||
http *http.Server
|
||||
dynamic *dynamicGateway
|
||||
auth *authorization
|
||||
}
|
||||
|
||||
func New(key string, expireSeconds int64, anonymous []string) (*Server, error) {
|
||||
auth, err := newAuthorization(key, expireSeconds, anonymous)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
grpcServer := grpc.NewServer(grpc.UnaryInterceptor(auth.unaryInterceptor))
|
||||
reflection.Register(grpcServer)
|
||||
engine := gin.New()
|
||||
engine.Use(gin.Logger(), gin.Recovery())
|
||||
return &Server{
|
||||
GRPC: grpcServer,
|
||||
Gateway: gwRuntime.NewServeMux(),
|
||||
HTTP: engine,
|
||||
auth: auth,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Server) Start(grpcAddr, httpAddr string) error {
|
||||
grpcListener, err := net.Listen("tcp", grpcAddr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listen gRPC on %s: %w", grpcAddr, err)
|
||||
}
|
||||
httpListener, err := net.Listen("tcp", httpAddr)
|
||||
if err != nil {
|
||||
_ = grpcListener.Close()
|
||||
return fmt.Errorf("listen HTTP on %s: %w", httpAddr, err)
|
||||
}
|
||||
|
||||
s.dynamic, err = newDynamicGateway(grpcAddr)
|
||||
if err != nil {
|
||||
_ = grpcListener.Close()
|
||||
_ = httpListener.Close()
|
||||
return err
|
||||
}
|
||||
s.HTTP.POST("/rpc/:module/:service/:method", s.dynamic.handle)
|
||||
|
||||
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
recorder := newBufferedResponse()
|
||||
s.Gateway.ServeHTTP(recorder, r)
|
||||
if recorder.status != http.StatusNotFound {
|
||||
recorder.flush(w)
|
||||
return
|
||||
}
|
||||
s.HTTP.ServeHTTP(w, r)
|
||||
})
|
||||
s.http = &http.Server{
|
||||
Addr: httpAddr,
|
||||
Handler: s.auth.httpMiddleware(handler),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
IdleTimeout: 120 * time.Second,
|
||||
MaxHeaderBytes: 1 << 20,
|
||||
}
|
||||
fmt.Printf("ecmall gRPC services listening on %s\n", grpcAddr)
|
||||
fmt.Printf("ecmall HTTP services listening on %s\n", httpAddr)
|
||||
|
||||
errCh := make(chan error, 2)
|
||||
go func() { errCh <- s.GRPC.Serve(grpcListener) }()
|
||||
go func() { errCh <- s.http.Serve(httpListener) }()
|
||||
serveErr := <-errCh
|
||||
if errors.Is(serveErr, grpc.ErrServerStopped) || errors.Is(serveErr, http.ErrServerClosed) {
|
||||
return http.ErrServerClosed
|
||||
}
|
||||
return serveErr
|
||||
}
|
||||
|
||||
func (s *Server) Stop(ctx context.Context) error {
|
||||
stopped := make(chan struct{})
|
||||
go func() {
|
||||
s.GRPC.GracefulStop()
|
||||
close(stopped)
|
||||
}()
|
||||
select {
|
||||
case <-stopped:
|
||||
case <-ctx.Done():
|
||||
s.GRPC.Stop()
|
||||
}
|
||||
if s.dynamic != nil {
|
||||
_ = s.dynamic.Close()
|
||||
}
|
||||
if s.http == nil {
|
||||
return nil
|
||||
}
|
||||
return s.http.Shutdown(ctx)
|
||||
}
|
||||
|
||||
type bufferedResponse struct {
|
||||
header http.Header
|
||||
body bytes.Buffer
|
||||
status int
|
||||
}
|
||||
|
||||
func newBufferedResponse() *bufferedResponse {
|
||||
return &bufferedResponse{header: make(http.Header), status: http.StatusOK}
|
||||
}
|
||||
|
||||
func (r *bufferedResponse) Header() http.Header { return r.header }
|
||||
func (r *bufferedResponse) WriteHeader(status int) { r.status = status }
|
||||
func (r *bufferedResponse) Write(data []byte) (int, error) { return r.body.Write(data) }
|
||||
func (r *bufferedResponse) flush(w http.ResponseWriter) {
|
||||
for key, values := range r.header {
|
||||
for _, value := range values {
|
||||
w.Header().Add(key, value)
|
||||
}
|
||||
}
|
||||
w.WriteHeader(r.status)
|
||||
_, _ = w.Write(r.body.Bytes())
|
||||
}
|
||||
24
pkgs/ecmall/internal/server/server_test.go
Normal file
24
pkgs/ecmall/internal/server/server_test.go
Normal file
@@ -0,0 +1,24 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func TestHTTPRouterIsAvailable(t *testing.T) {
|
||||
srv, err := New("0123456789abcdef0123456789abcdef", 3600, []string{"/healthz"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
srv.HTTP.GET("/healthz", func(c *gin.Context) { c.Status(http.StatusNoContent) })
|
||||
|
||||
request := httptest.NewRequest(http.MethodGet, "/healthz", nil)
|
||||
response := httptest.NewRecorder()
|
||||
srv.HTTP.ServeHTTP(response, request)
|
||||
if response.Code != http.StatusNoContent {
|
||||
t.Fatalf("unexpected status: %d", response.Code)
|
||||
}
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/address.go
Normal file
20
pkgs/ecmall/internal/service/address.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/ec/address/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeAddress(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/ads.go
Normal file
20
pkgs/ecmall/internal/service/ads.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/ads/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeAds(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/cms.go
Normal file
20
pkgs/ecmall/internal/service/cms.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/cms/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeCMS(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/feedback.go
Normal file
20
pkgs/ecmall/internal/service/feedback.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/feedback/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeFeedback(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
16
pkgs/ecmall/internal/service/fts.go
Normal file
16
pkgs/ecmall/internal/service/fts.go
Normal file
@@ -0,0 +1,16 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/fts/service"
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeFTS(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{Redis: impl.RedisService, Etcd: impl.EtcdService, DB: impl.DBService, Cache: impl.MemoryService},
|
||||
Engine: srv.HTTP,
|
||||
Config: config.Spec.Fts,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/initial.go
Normal file
20
pkgs/ecmall/internal/service/initial.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/initial/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeInitial(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
14
pkgs/ecmall/internal/service/logs.go
Normal file
14
pkgs/ecmall/internal/service/logs.go
Normal file
@@ -0,0 +1,14 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/logs/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeLogs(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{Redis: impl.RedisService, Etcd: impl.EtcdService, DB: impl.DBService, Cache: impl.MemoryService},
|
||||
Engine: srv.HTTP,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/mall.go
Normal file
20
pkgs/ecmall/internal/service/mall.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/ec/mall/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeMall(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/market.go
Normal file
20
pkgs/ecmall/internal/service/market.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/ec/market/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeMarket(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
16
pkgs/ecmall/internal/service/mgt.go
Normal file
16
pkgs/ecmall/internal/service/mgt.go
Normal file
@@ -0,0 +1,16 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/mgt/service"
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeMgt(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{Redis: impl.RedisService, Etcd: impl.EtcdService, DB: impl.DBService, Cache: impl.MemoryService},
|
||||
Engine: srv.HTTP,
|
||||
Config: config.Spec.Mgt,
|
||||
})
|
||||
}
|
||||
20
pkgs/ecmall/internal/service/order.go
Normal file
20
pkgs/ecmall/internal/service/order.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/ec/order/service"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeOrder(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
})
|
||||
}
|
||||
22
pkgs/ecmall/internal/service/passport.go
Normal file
22
pkgs/ecmall/internal/service/passport.go
Normal file
@@ -0,0 +1,22 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/passport/service"
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposePassport(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
Config: config.Spec.Passport,
|
||||
})
|
||||
}
|
||||
22
pkgs/ecmall/internal/service/sender.go
Normal file
22
pkgs/ecmall/internal/service/sender.go
Normal file
@@ -0,0 +1,22 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/base/sender/service"
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeSender(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
Config: config.Spec.Sender,
|
||||
})
|
||||
}
|
||||
45
pkgs/ecmall/internal/service/service.go
Normal file
45
pkgs/ecmall/internal/service/service.go
Normal file
@@ -0,0 +1,45 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
"github.com/gin-contrib/sessions"
|
||||
"github.com/gin-contrib/sessions/cookie"
|
||||
)
|
||||
|
||||
type exposeFunc func(*server.Server) error
|
||||
|
||||
var services = []struct {
|
||||
name string
|
||||
expose exposeFunc
|
||||
}{
|
||||
{"ads", exposeAds},
|
||||
{"cms", exposeCMS},
|
||||
{"feedback", exposeFeedback},
|
||||
{"fts", exposeFTS},
|
||||
{"initial", exposeInitial},
|
||||
{"logs", exposeLogs},
|
||||
{"mgt", exposeMgt},
|
||||
{"passport", exposePassport},
|
||||
{"sender", exposeSender},
|
||||
{"address", exposeAddress},
|
||||
{"mall", exposeMall},
|
||||
{"market", exposeMarket},
|
||||
{"order", exposeOrder},
|
||||
{"wallet", exposeWallet},
|
||||
}
|
||||
|
||||
func Expose(srv *server.Server) error {
|
||||
srv.HTTP.Use(sessions.Sessions("mysession", cookie.NewStore([]byte(config.Spec.SecretKey+"-session"))))
|
||||
for _, item := range services {
|
||||
if !config.Enabled(item.name) {
|
||||
continue
|
||||
}
|
||||
if err := item.expose(srv); err != nil {
|
||||
return fmt.Errorf("expose %s: %w", item.name, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
62
pkgs/ecmall/internal/service/service_test.go
Normal file
62
pkgs/ecmall/internal/service/service_test.go
Normal file
@@ -0,0 +1,62 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
ecmallserver "bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
// TestExposeRegistersPlatformModules guards the aggregate registry: the
|
||||
// platform modules (ads, cms, feedback, fts, logs, mgt) must contribute their
|
||||
// gRPC services and their native REST routes through service.Expose, and the
|
||||
// gRPC-only modules must keep contributing theirs.
|
||||
func TestExposeRegistersPlatformModules(t *testing.T) {
|
||||
t.Setenv("BSM_SERVICES", "")
|
||||
config.Spec.Services = []string{"all"}
|
||||
|
||||
srv, err := ecmallserver.New("0123456789abcdef0123456789abcdef", 3600, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("new server: %v", err)
|
||||
}
|
||||
if err := Expose(srv); err != nil {
|
||||
t.Fatalf("expose: %v", err)
|
||||
}
|
||||
|
||||
services := make(map[string]bool)
|
||||
for name := range srv.GRPC.GetServiceInfo() {
|
||||
services[name] = true
|
||||
}
|
||||
for _, prefix := range []string{"ads.", "cms.", "feedback.", "address.", "mall.", "market.", "order.", "passport.", "sender.", "wallet.", "initial."} {
|
||||
if !hasPrefix(services, prefix) {
|
||||
t.Errorf("gRPC service with prefix %q is not registered", prefix)
|
||||
}
|
||||
}
|
||||
|
||||
routes := make(map[string]bool)
|
||||
for _, route := range srv.HTTP.Routes() {
|
||||
routes[route.Method+" "+route.Path] = true
|
||||
}
|
||||
for _, path := range []string{
|
||||
"GET /rest/fts/ping",
|
||||
"GET /rest/fts/config",
|
||||
"POST /rest/fts/uploader",
|
||||
"GET /rest/logs/ping",
|
||||
"GET /rest/mgt/ping",
|
||||
"POST /rest/mgt/login",
|
||||
} {
|
||||
if !routes[path] {
|
||||
t.Errorf("native REST route %q is not registered", path)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func hasPrefix(names map[string]bool, prefix string) bool {
|
||||
for name := range names {
|
||||
if strings.HasPrefix(name, prefix) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
22
pkgs/ecmall/internal/service/wallet.go
Normal file
22
pkgs/ecmall/internal/service/wallet.go
Normal file
@@ -0,0 +1,22 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
moduleService "bsm/full/module/finance/wallet/service"
|
||||
"bsm/full/pkgs/ecmall/internal/config"
|
||||
"bsm/full/pkgs/ecmall/internal/impl"
|
||||
"bsm/full/pkgs/ecmall/internal/server"
|
||||
)
|
||||
|
||||
func exposeWallet(srv *server.Server) error {
|
||||
return moduleService.Expose(moduleService.ExposeOptions{
|
||||
Dependencies: moduleService.Dependencies{
|
||||
Redis: impl.RedisService,
|
||||
Etcd: impl.EtcdService,
|
||||
DB: impl.DBService,
|
||||
Cache: impl.MemoryService,
|
||||
},
|
||||
GRPC: srv.GRPC,
|
||||
Gateway: srv.Gateway,
|
||||
Config: config.Spec.Wallet,
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user