Compare commits
8 Commits
Author | SHA1 | Date | |
---|---|---|---|
46891c397f | |||
6856038abe | |||
786bbb7185 | |||
95207c9617 | |||
d646deb468 | |||
468819f0a0 | |||
832f1034a8 | |||
f0b6370ee1 |
18
go.mod
18
go.mod
@@ -1,11 +1,17 @@
|
||||
module go.unistack.org/micro-server-grpc/v3
|
||||
|
||||
go 1.16
|
||||
go 1.20
|
||||
|
||||
require (
|
||||
github.com/golang/protobuf v1.5.2
|
||||
go.unistack.org/micro/v3 v3.10.14
|
||||
golang.org/x/net v0.5.0
|
||||
google.golang.org/grpc v1.52.3
|
||||
google.golang.org/protobuf v1.28.1
|
||||
github.com/golang/protobuf v1.5.3
|
||||
go.unistack.org/micro/v3 v3.10.31
|
||||
golang.org/x/net v0.17.0
|
||||
google.golang.org/grpc v1.59.0
|
||||
google.golang.org/protobuf v1.31.0
|
||||
)
|
||||
|
||||
require (
|
||||
golang.org/x/sys v0.13.0 // indirect
|
||||
golang.org/x/text v0.13.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231030173426-d783a09b4405 // indirect
|
||||
)
|
||||
|
84
grpc.go
84
grpc.go
@@ -26,7 +26,6 @@ import (
|
||||
"golang.org/x/net/netutil"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/encoding"
|
||||
gmetadata "google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/peer"
|
||||
@@ -128,6 +127,10 @@ func (g *Server) configure(opts ...server.Option) error {
|
||||
}
|
||||
}
|
||||
|
||||
for _, k := range g.opts.Codecs {
|
||||
encoding.RegisterCodec(&wrapMicroCodec{k})
|
||||
}
|
||||
|
||||
maxMsgSize := g.getMaxMsgSize()
|
||||
|
||||
gopts := []grpc.ServerOption{
|
||||
@@ -136,10 +139,6 @@ func (g *Server) configure(opts ...server.Option) error {
|
||||
grpc.UnknownServiceHandler(g.handler),
|
||||
}
|
||||
|
||||
if creds := g.getCredentials(); creds != nil {
|
||||
gopts = append(gopts, grpc.Creds(creds))
|
||||
}
|
||||
|
||||
if opts := g.getGrpcOptions(); opts != nil {
|
||||
gopts = append(opts, gopts...)
|
||||
}
|
||||
@@ -176,13 +175,6 @@ func (g *Server) getMaxMsgSize() int {
|
||||
return s
|
||||
}
|
||||
|
||||
func (g *Server) getCredentials() credentials.TransportCredentials {
|
||||
if g.opts.TLSConfig != nil {
|
||||
return credentials.NewTLS(g.opts.TLSConfig)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (g *Server) getGrpcOptions() []grpc.ServerOption {
|
||||
if g.opts.Context == nil {
|
||||
return nil
|
||||
@@ -242,6 +234,10 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) (err error)
|
||||
for k, v := range gmd {
|
||||
md.Set(k, strings.Join(v, ", "))
|
||||
}
|
||||
md.Set("Path", fullMethod)
|
||||
md.Set("Micro-Server", "grpc")
|
||||
md.Set(metadata.HeaderEndpoint, methodName)
|
||||
md.Set(metadata.HeaderService, serviceName)
|
||||
|
||||
var td string
|
||||
// timeout for server deadline
|
||||
@@ -714,31 +710,6 @@ func (g *Server) Register() error {
|
||||
g.Lock()
|
||||
defer g.Unlock()
|
||||
|
||||
for sb := range g.subscribers {
|
||||
handler := g.createSubHandler(sb, config)
|
||||
var opts []broker.SubscribeOption
|
||||
if queue := sb.Options().Queue; len(queue) > 0 {
|
||||
opts = append(opts, broker.SubscribeGroup(queue))
|
||||
}
|
||||
|
||||
subCtx := config.Context
|
||||
if cx := sb.Options().Context; cx != nil {
|
||||
subCtx = cx
|
||||
}
|
||||
opts = append(opts, broker.SubscribeContext(subCtx))
|
||||
opts = append(opts, broker.SubscribeAutoAck(sb.Options().AutoAck))
|
||||
opts = append(opts, broker.SubscribeBodyOnly(sb.Options().BodyOnly))
|
||||
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Subscribing to topic: %s", sb.Topic())
|
||||
}
|
||||
sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), handler, opts...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g.subscribers[sb] = []broker.Subscriber{sub}
|
||||
}
|
||||
|
||||
g.registered = true
|
||||
g.rsvc = service
|
||||
|
||||
@@ -809,10 +780,6 @@ func (g *Server) Start() error {
|
||||
|
||||
config := g.Options()
|
||||
|
||||
for _, k := range config.Codecs {
|
||||
encoding.RegisterCodec(&wrapMicroCodec{k})
|
||||
}
|
||||
|
||||
// micro: config.Transport.Listen(config.Address)
|
||||
var ts net.Listener
|
||||
var err error
|
||||
@@ -876,6 +843,10 @@ func (g *Server) Start() error {
|
||||
}
|
||||
}
|
||||
|
||||
if err = g.subscribe(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// micro: go ts.Accept(s.accept)
|
||||
go func() {
|
||||
if err = g.srv.Serve(ts); err != nil {
|
||||
@@ -987,6 +958,37 @@ func (g *Server) Start() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (g *Server) subscribe() error {
|
||||
config := g.opts
|
||||
|
||||
for sb := range g.subscribers {
|
||||
handler := g.createSubHandler(sb, config)
|
||||
var opts []broker.SubscribeOption
|
||||
if queue := sb.Options().Queue; len(queue) > 0 {
|
||||
opts = append(opts, broker.SubscribeGroup(queue))
|
||||
}
|
||||
|
||||
subCtx := config.Context
|
||||
if cx := sb.Options().Context; cx != nil {
|
||||
subCtx = cx
|
||||
}
|
||||
opts = append(opts, broker.SubscribeContext(subCtx))
|
||||
opts = append(opts, broker.SubscribeAutoAck(sb.Options().AutoAck))
|
||||
opts = append(opts, broker.SubscribeBodyOnly(sb.Options().BodyOnly))
|
||||
|
||||
if config.Logger.V(logger.InfoLevel) {
|
||||
config.Logger.Infof(config.Context, "Subscribing to topic: %s", sb.Topic())
|
||||
}
|
||||
sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), handler, opts...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g.subscribers[sb] = []broker.Subscriber{sub}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (g *Server) Stop() error {
|
||||
g.RLock()
|
||||
if !g.started {
|
||||
|
Reference in New Issue
Block a user