You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
72 lines
2.3 KiB
72 lines
2.3 KiB
package generic |
|
|
|
import ( |
|
"context" |
|
"fmt" |
|
"sig-pub/pkg/utils/collect" |
|
"sig-pub/pkg/utils/retry" |
|
"sig-pub/pkg/zlog" |
|
"time" |
|
|
|
"google.golang.org/grpc" |
|
) |
|
|
|
type GrpcGenericClientFactory struct { |
|
scheme string |
|
defaultOpts []grpc.DialOption |
|
clientCache *collect.ConcurrentMap[string, *GrpcGenericClient] |
|
} |
|
|
|
func NewGrpcGenericClientFactory(scheme string, defaultOpts ...grpc.DialOption) *GrpcGenericClientFactory { |
|
return &GrpcGenericClientFactory{ |
|
scheme: scheme, |
|
defaultOpts: defaultOpts, |
|
} |
|
} |
|
|
|
func (f *GrpcGenericClientFactory) Init() (err error) { |
|
f.clientCache = collect.NewConcurrentMap[string, *GrpcGenericClient](8, func(serviceName string) string { return serviceName }) |
|
return |
|
} |
|
|
|
func (f *GrpcGenericClientFactory) NewClient(ctx context.Context, serviceName string, opts ...grpc.DialOption) (client *GrpcGenericClient, err error) { |
|
addr := fmt.Sprintf("%s:///%s", f.scheme, serviceName) |
|
dialOpts := make([]grpc.DialOption, 0, len(f.defaultOpts)+len(opts)) |
|
dialOpts = append(dialOpts, f.defaultOpts...) |
|
dialOpts = append(dialOpts, opts...) |
|
conn, err := grpc.NewClient(addr, dialOpts...) |
|
if err != nil { |
|
return |
|
} |
|
client = NewGpcGenericClient(serviceName, conn) |
|
err = client.InitStub(ctx) |
|
return |
|
} |
|
|
|
func (f *GrpcGenericClientFactory) GetClient(ctx context.Context, serviceName string, opts ...grpc.DialOption) (client *GrpcGenericClient, err error) { |
|
// todo 优化 |
|
client, err, _ = f.clientCache.ComputeIfAbsentE(serviceName, func(serviceName string) (*GrpcGenericClient, error) { |
|
return f.NewClient(ctx, serviceName, opts...) |
|
}) |
|
return |
|
} |
|
|
|
// RefreshService 服务重新注册时可能有更改, 刷新旧的 grpc generic stub |
|
func (f *GrpcGenericClientFactory) RefreshService(serviceName string, passing bool) { |
|
zlog.Debugf("service status updated: %s, health=%v", serviceName, passing) |
|
if passing { |
|
service, ok := f.clientCache.Load(serviceName) |
|
if ok { |
|
go retry.DoWithFixDelay(10, 2*time.Second, func(_ uint32) (_ struct{}, err error) { |
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) |
|
defer cancel() |
|
if err = service.InitStub(ctx); err != nil { |
|
zlog.Errorf("refresh service stub error: %s, %v", serviceName, err) |
|
} else { |
|
zlog.Debugf("refresh service stub success: %s", serviceName) |
|
} |
|
return |
|
}) |
|
} |
|
} |
|
}
|
|
|