4
0

Compare commits

...

8 Commits
v1.3.8 ... main

Author SHA1 Message Date
dc.To
6ed8f159e7 修改依赖路径 2025-08-27 15:03:16 +08:00
dc.To
c9133edfd9 update dependency library 2025-08-07 10:16:58 +08:00
dc.To
fed21bfb86 update etcd variable 2025-08-07 10:03:10 +08:00
dc.To
981e1e0cf0 update invoke add atomic with context config 2025-08-07 09:52:32 +08:00
dc.To
7db81a88e4 添加配置调用 2025-08-06 17:51:00 +08:00
dc.To
ffdfa9c94f remove debug code 2025-07-10 11:10:17 +08:00
dc.To
fe25f60b5d 更新超时链 2025-07-10 11:02:46 +08:00
dc.To
ccf5df87e7 update timeout invoker 2025-06-27 14:37:24 +08:00
4 changed files with 132 additions and 22 deletions

70
apapter/atomic.go Normal file
View File

@ -0,0 +1,70 @@
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
}

50
apapter/context.go Normal file
View File

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

View File

@ -4,20 +4,20 @@ import (
"os"
"path/filepath"
"git.linkiio.cn/linkpay/invoke/chain"
"git.linkiio.cn/linkpay/protobuf"
"git.blueorigin.work/blueorigin/invoke/chain"
"git.blueorigin.work/blueorigin/protobuf"
"github.com/gogf/gf/contrib/registry/etcd/v2"
"github.com/gogf/gf/contrib/rpc/grpcx/v2"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
)
var Endpoint = os.Getenv("ETCD_ENDPOINTS")
var Endpoints = os.Getenv("ETCD_ENDPOINTS")
var GrpcConfig *grpcx.GrpcServerConfig = grpcx.Server.NewConfig()
func Etcd() *etcd.Registry {
return etcd.New(Endpoint)
return etcd.New(Endpoints)
}
func SetOptions(o ...grpc.ServerOption) *grpcx.GrpcServerConfig {
@ -40,7 +40,7 @@ func Registry(f ...func(s *grpcx.GrpcServer)) *grpcx.GrpcServer {
chain.ClientMetadataUnaryChain,
)}...,
)
if Endpoint != "" {
if Endpoints != "" {
grpcx.Resolver.Register(Etcd())
}
@ -65,7 +65,7 @@ func Registry(f ...func(s *grpcx.GrpcServer)) *grpcx.GrpcServer {
* @version 20250427
*/
func Reflect(s reflection.GRPCServer, f ...func()) {
if Endpoint == "" {
if Endpoints == "" {
reflection.Register(s)
}
for _, fn := range f {