prepare v4 #139

Merged
vtolstov merged 10 commits from devstigneev/micro-client-grpc:prepare_v4 into v4 2025-03-03 09:53:41 +03:00
Showing only changes of commit 1f8a64c7d6 - Show all commits

68
grpc.go
View File

@@ -693,74 +693,6 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
return nil, grr return nil, grr
} }
/*
func (g *grpcClient) BatchPublish(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.publish(ctx, []client.Message{p}, opts...)
}
func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
var body []byte
options := client.NewPublishOptions(opts...)
// get proxy
exchange := ""
if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
exchange = v
}
msgs := make([]*broker.Message, 0, len(ps))
omd, ok := metadata.FromOutgoingContext(ctx)
if !ok {
omd = metadata.New(2)
}
for _, p := range ps {
md := metadata.Copy(omd)
md[metadata.HeaderContentType] = p.ContentType()
// passed in raw data
if d, ok := p.Payload().(*codec.Frame); ok {
body = d.Data
} else {
// use codec for payload
cf, err := g.newCodec(p.ContentType())
if err != nil {
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", err.Error())
}
body = b
}
topic := p.Topic()
if len(exchange) > 0 {
topic = exchange
}
for k, v := range p.Metadata() {
md.Set(k, v)
}
md.Set(metadata.HeaderTopic, topic)
msgs = append(msgs, &broker.Message{Header: md, Body: body})
}
return g.opts.Broker.BatchPublish(ctx, msgs,
broker.PublishContext(options.Context),
broker.PublishBodyOnly(options.BodyOnly),
)
}
*/
func (g *grpcClient) String() string { func (g *grpcClient) String() string {
return "grpc" return "grpc"
} }