unistack-org/micro-client-grpc
4
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases 40 Wiki Activity

prepare v4

Browse Source
This commit is contained in:
Денис Евстигнеев
2025-02-25 16:42:21 +03:00
parent cee07d11d4
commit 5a9b5e23fc
9 changed files with 111 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
grpc.go
View File

@@ -6,22 +6,17 @@ import (
"crypto/tls"
"fmt"
"net"
"os"
"reflect"
"strconv"
"strings"
"sync"
"time"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/codec"
"go.unistack.org/micro/v3/errors"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/options"
"go.unistack.org/micro/v3/selector"
"go.unistack.org/micro/v3/semconv"
"go.unistack.org/micro/v3/tracer"
"go.unistack.org/micro/v4/client"
"go.unistack.org/micro/v4/codec"
"go.unistack.org/micro/v4/errors"
"go.unistack.org/micro/v4/metadata"
"go.unistack.org/micro/v4/selector"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
@@ -78,15 +73,15 @@ func (g *grpcClient) secure(addr string) grpc.DialOption {
}
func (g *grpcClient) call(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string
var header map[string][]string
if md, ok := metadata.FromOutgoingContext(ctx); ok {
header = make(map[string]string, len(md))
header = make(map[string][]string, len(md))
for k, v := range md {
header[strings.ToLower(k)] = v
}
} else {
header = make(map[string]string, 2)
header = make(map[string][]string, 2)
}
if opts.RequestMetadata != nil {
for k, v := range opts.RequestMetadata {
@@ -94,11 +89,11 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
}
}
// set timeout in nanoseconds
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
header["timeout"] = fmt.Sprintf("%dn", opts.RequestTimeout)
header["content-type"] = req.ContentType()
header["Grpc-Timeout"] = append(header["Grpc-Timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
header["timeout"] = append(header["timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
header["content-type"] = append(header["content-type"], req.ContentType())
md := gmetadata.New(header)
md := gmetadata.MD(metadata.Copy(header))
ctx = gmetadata.NewOutgoingContext(ctx, md)
cf, err := g.newCodec(req.ContentType())
@@ -193,26 +188,26 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
}
func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string
var header map[string][]string
if md, ok := metadata.FromOutgoingContext(ctx); ok {
header = make(map[string]string, len(md))
header = make(map[string][]string, len(md))
for k, v := range md {
header[k] = v
}
} else {
header = make(map[string]string)
header = make(map[string][]string)
}
// set timeout in nanoseconds
if opts.StreamTimeout > time.Duration(0) {
header["Grpc-Timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
header["timeout"] = fmt.Sprintf("%dn", opts.StreamTimeout)
header["Grpc-Timeout"] = append(header["Grpc-Timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
header["timeout"] = append(header["timeout"], fmt.Sprintf("%dn", opts.RequestTimeout))
}
// set the content type for the request
header["content-type"] = req.ContentType()
header["content-type"] = append(header["content-type"], req.ContentType())
md := gmetadata.New(header)
md := gmetadata.MD(metadata.Copy(header))
ctx = gmetadata.NewOutgoingContext(ctx, md)
cf, err := g.newCodec(req.ContentType())
@@ -441,10 +436,6 @@ func (g *grpcClient) Options() client.Options {
return g.opts
}
func (g *grpcClient) NewMessage(topic string, msg interface{}, opts ...client.MessageOption) client.Message {
return newGRPCEvent(topic, msg, g.opts.ContentType, opts...)
}
func (g *grpcClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
return newGRPCRequest(service, method, req, g.opts.ContentType, reqOpts...)
}
@@ -763,19 +754,12 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
return nil, grr
}
/*
func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
return g.funcBatchPublish(ctx, ps, opts...)
}
func (g *grpcClient) fnBatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
return g.publish(ctx, ps, opts...)
}
func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
return g.funcPublish(ctx, p, opts...)
}
func (g *grpcClient) fnPublish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
return g.publish(ctx, []client.Message{p}, opts...)
}
@@ -789,10 +773,6 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
exchange = v
}
// get the exchange
if len(options.Exchange) > 0 {
exchange = options.Exchange
}
msgs := make([]*broker.Message, 0, len(ps))
@@ -803,17 +783,6 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
for _, p := range ps {
md := metadata.Copy(omd)
topic := p.Topic()
if len(exchange) > 0 {
topic = exchange
}
md.Set(metadata.HeaderTopic, topic)
iter := p.Metadata().Iterator()
var k, v string
for iter.Next(&k, &v) {
md.Set(k, v)
}
md[metadata.HeaderContentType] = p.ContentType()
// passed in raw data
@@ -823,12 +792,12 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
// use codec for payload
cf, err := g.newCodec(p.ContentType())
if err != nil {
return errors.InternalServerError("go.micro.client", "%+v", err)
return errors.InternalServerError("go.micro.client", err.Error())
}
// set the body
b, err := cf.Marshal(p.Payload())
if err != nil {
return errors.InternalServerError("go.micro.client", "%+v", err)
return errors.InternalServerError("go.micro.client", err.Error())
}
body = b
}
@@ -841,6 +810,8 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
)
}
*/
func (g *grpcClient) String() string {
return "grpc"
}