4
0

Compare commits

..

No commits in common. "main" and "v1.3.5" have entirely different histories.
main ... v1.3.5

4 changed files with 27 additions and 145 deletions

View File

@ -1,70 +0,0 @@
package apapter
import (
"context"
"errors"
"sync/atomic"
"github.com/gogf/gf/v2/container/gvar"
"github.com/gogf/gf/v2/encoding/gbase64"
"github.com/gogf/gf/v2/encoding/gjson"
"github.com/gogf/gf/v2/frame/g"
"google.golang.org/grpc/metadata"
)
/**
* 原子钟接口配置
* @author dc.To
* @version 20250807
*/
type AtomicAdapter struct {
store atomic.Value
adapter string
}
func NewAtomicAdapter(ctx context.Context, adapter string) *AtomicAdapter {
a := &AtomicAdapter{adapter: adapter}
md, ok := metadata.FromIncomingContext(ctx)
if ok {
if c := md.Get(a.adapter); len(c) > 0 {
a.SetContent(gjson.New(gbase64.MustDecodeToString(c[0])).Map())
} else {
g.Log("cfg").Error(ctx, "No Metadata from incoming cfg context")
}
}
return a
}
func (a *AtomicAdapter) SetContent(content map[string]interface{}) error {
a.store.Store(content)
return nil
}
func (a *AtomicAdapter) Available(ctx context.Context, resource ...string) (ok bool) {
return a.store.Load() != nil
}
func (a *AtomicAdapter) Get(ctx context.Context, pattern string) (value interface{}, err error) {
raw := a.store.Load()
if raw == nil {
return nil, errors.New("config not loaded")
}
data := raw.(map[string]interface{})
if v, ok := data[pattern]; ok {
return gvar.New(v), nil
}
return gvar.New(nil), nil
}
func (a *AtomicAdapter) Data(ctx context.Context) (data map[string]interface{}, err error) {
raw := a.store.Load()
if raw == nil {
return nil, errors.New("config not loaded")
}
original := raw.(map[string]interface{})
clone := make(map[string]interface{}, len(original))
for k, v := range original {
clone[k] = v
}
return clone, nil
}

View File

@ -1,50 +0,0 @@
package apapter
import (
"context"
"github.com/gogf/gf/v2/container/gvar"
"github.com/gogf/gf/v2/encoding/gbase64"
"github.com/gogf/gf/v2/encoding/gjson"
"github.com/gogf/gf/v2/frame/g"
"google.golang.org/grpc/metadata"
)
/**
* @Description: ContextAdapter
* @author dc.To
* @version 20250807
*/
type ContextAdapter struct {
ctx context.Context
adapter string
}
func NewContextAdapter(ctx context.Context, adapter string) *ContextAdapter {
return &ContextAdapter{ctx: ctx, adapter: adapter}
}
func (a *ContextAdapter) ctxWithConfig(ctx context.Context) *gvar.Var {
md, ok := metadata.FromIncomingContext(ctx)
if ok {
if c := md.Get(a.adapter); len(c) > 0 {
return gvar.New(gbase64.MustDecodeToString(c[0]))
}
}
g.Log("cfg").Error(ctx, "No Metadata from ["+a.adapter+"] cfg context")
return gvar.New(nil)
}
func (a *ContextAdapter) Available(ctx context.Context, resource ...string) (ok bool) {
return a.ctxWithConfig(ctx) != nil
}
func (a *ContextAdapter) Get(ctx context.Context, pattern string) (value interface{}, err error) {
return gjson.New(a.ctxWithConfig(ctx)).Get(pattern).Val(), nil
}
func (a *ContextAdapter) Data(ctx context.Context) (data map[string]interface{}, err error) {
return gjson.New(a.ctxWithConfig(ctx)).Map(), nil
}

View File

@ -3,9 +3,11 @@ package chain
import ( import (
"context" "context"
"strings" "strings"
"time"
"git.blueorigin.work/blueorigin/invoke/apapter" "github.com/gogf/gf/v2/encoding/gbase64"
"github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gcfg"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
) )
@ -16,16 +18,19 @@ import (
* @version 20250409 * @version 20250409
*/ */
func ClientTimeoutInvokerChain(ctx context.Context, method string, req, res interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { func ClientTimeoutInvokerChain(ctx context.Context, method string, req, res interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
return invoker(ctx, method, req, res, cc, opts...) return invoker(ctx, method, req, res, cc, opts...)
} }
/** /**
* 上下文调用 * 上下文拦截
* @param void * @param void
* @author dc.To * @author dc.To
* @version 20250424 * @version 20250424
*/ */
func ClientContextInvokerChain(ctx context.Context, method string, req, res interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { func ClientContextInvokerChain(ctx context.Context, method string, req, res interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
return invoker(ctx, method, req, nil, cc, opts...) return invoker(ctx, method, req, nil, cc, opts...)
} }
@ -36,6 +41,7 @@ func ClientContextInvokerChain(ctx context.Context, method string, req, res inte
* @version 20250425 * @version 20250425
*/ */
func ClientContextUnaryChain(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { func ClientContextUnaryChain(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
return handler(ctx, req) return handler(ctx, req)
} }
@ -45,16 +51,20 @@ func ClientContextUnaryChain(ctx context.Context, req interface{}, info *grpc.Un
* @version 20250418 * @version 20250418
*/ */
func ClientMetadataUnaryChain(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { func ClientMetadataUnaryChain(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
md, ok := metadata.FromIncomingContext(ctx) md, ok := metadata.FromIncomingContext(ctx)
if ok { if ok {
for k, v := range md { for k, v := range md {
if strings.HasPrefix(k, "x-cfg-") && len(v) > 0 { if strings.HasPrefix(k, "x-cfg-") && len(v) > 0 {
g.Cfg(strings.TrimPrefix(k, "x-cfg-")).SetAdapter(apapter.NewContextAdapter(ctx, k)) c := strings.Join(v, "")
if len(c) > 0 {
adapter, err := gcfg.NewAdapterContent(gbase64.MustDecodeToString(v[0]))
if err != nil {
g.Log("unary").Error(ctx, err)
}
g.Cfg(strings.TrimPrefix(k, "x-cfg-")).SetAdapter(adapter)
}
} }
} }
} else {
g.Log("unary").Error(ctx, "No Metadata from incoming cfg context")
} }
return handler(ctx, req) return handler(ctx, req)
} }

View File

@ -2,22 +2,20 @@ package grpcx
import ( import (
"os" "os"
"path/filepath"
"git.blueorigin.work/blueorigin/invoke/chain" "git.linkiio.cn/linkpay/invoke/chain"
"git.blueorigin.work/blueorigin/protobuf"
"github.com/gogf/gf/contrib/registry/etcd/v2" "github.com/gogf/gf/contrib/registry/etcd/v2"
"github.com/gogf/gf/contrib/rpc/grpcx/v2" "github.com/gogf/gf/contrib/rpc/grpcx/v2"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/reflection" "google.golang.org/grpc/reflection"
) )
var Endpoints = os.Getenv("ETCD_ENDPOINTS") var Endpoint = os.Getenv("ETCD_ENDPOINTS")
var GrpcConfig *grpcx.GrpcServerConfig = grpcx.Server.NewConfig() var GrpcConfig *grpcx.GrpcServerConfig = grpcx.Server.NewConfig()
func Etcd() *etcd.Registry { func Etcd() *etcd.Registry {
return etcd.New(Endpoints) return etcd.New(Endpoint)
} }
func SetOptions(o ...grpc.ServerOption) *grpcx.GrpcServerConfig { func SetOptions(o ...grpc.ServerOption) *grpcx.GrpcServerConfig {
@ -40,14 +38,10 @@ func Registry(f ...func(s *grpcx.GrpcServer)) *grpcx.GrpcServer {
chain.ClientMetadataUnaryChain, chain.ClientMetadataUnaryChain,
)}..., )}...,
) )
if Endpoints != "" { if Endpoint != "" {
grpcx.Resolver.Register(Etcd()) grpcx.Resolver.Register(Etcd())
} }
if GrpcConfig.Name == "default" || GrpcConfig.Name == "" {
GrpcConfig.Name = filepath.Base(os.Args[0])
}
s := grpcx.Server.New(GrpcConfig) s := grpcx.Server.New(GrpcConfig)
for _, fn := range f { for _, fn := range f {
@ -65,7 +59,7 @@ func Registry(f ...func(s *grpcx.GrpcServer)) *grpcx.GrpcServer {
* @version 20250427 * @version 20250427
*/ */
func Reflect(s reflection.GRPCServer, f ...func()) { func Reflect(s reflection.GRPCServer, f ...func()) {
if Endpoints == "" { if Endpoint == "" {
reflection.Register(s) reflection.Register(s)
} }
for _, fn := range f { for _, fn := range f {
@ -73,11 +67,9 @@ func Reflect(s reflection.GRPCServer, f ...func()) {
} }
} }
/** // func Resolve[T any](value T, f func(T) T) T {
* BSS 双向通信流 // if f == nil {
* @author dc.To // return value
* @version 20250410 // }
*/ // return f(value)
func BssClient() protobuf.BssClient { // }
return protobuf.NewBssClient(grpcx.Client.MustNewGrpcClientConn("api"))
}