go/kratos
1
0
Fork 0
mirror of https://github.com/go-kratos/kratos.git synced 2025-03-17 21:07:54 +02:00
Code Issues Releases Activity

add feat endpoint (#972)

Browse Source 
* add feature endpoint
This commit is contained in:
longxboy 2021-05-31 16:10:17 +08:00 committed by GitHub
parent 493373dc13
commit e1d6377542
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 54 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
app.go
View File

@ -116,22 +116,27 @@ func (a *App) Stop() error {
}
func (a *App) buildInstance() (*registry.ServiceInstance, error) {
if len(a.opts.endpoints) == 0 {
var endpoints []string
for _, e := range a.opts.endpoints {
endpoints = append(endpoints, e.String())
}
if len(endpoints) == 0 {
for _, srv := range a.opts.servers {
if r, ok := srv.(transport.Endpointer); ok {
e, err := r.Endpoint()
if err != nil {
return nil, err
}
a.opts.endpoints = append(a.opts.endpoints, e)
endpoints = append(endpoints, e.String())
}
}
}
return &registry.ServiceInstance{
ID: a.opts.id,
Name: a.opts.name,
Version: a.opts.version,
Metadata: a.opts.metadata,
Endpoints: a.opts.endpoints,
Endpoints: endpoints,
}, nil
}

5
options.go
View File

@ -2,6 +2,7 @@ package kratos
import (
"context"
"net/url"
"os"
"github.com/go-kratos/kratos/v2/log"
@ -18,7 +19,7 @@ type options struct {
name string
version string
metadata map[string]string
endpoints []string
endpoints []*url.URL
ctx context.Context
sigs []os.Signal
@ -50,7 +51,7 @@ func Metadata(md map[string]string) Option {
}
// Endpoint with service endpoint.
func Endpoint(endpoints ...string) Option {
func Endpoint(endpoints ...*url.URL) Option {
return func(o *options) { o.endpoints = endpoints }
}

7
transport/grpc/context.go
View File

@ -1,6 +1,9 @@
package grpc
import "context"
import (
"context"
"net/url"
)
// ServerInfo represent gRPC server information.
type ServerInfo struct {
@ -8,6 +11,8 @@ type ServerInfo struct {
Server interface{}
// FullMethod is the full RPC method string, i.e., /package.service/method.
FullMethod string
// Endpoint is a real address to registry endpoint.
Endpoint *url.URL
}
type serverKey struct{}

18
transport/grpc/server.go
View File

@ -2,8 +2,8 @@ package grpc
import (
"context"
"fmt"
"net"
"net/url"
"strings"
"time"
@ -75,6 +75,7 @@ type Server struct {
lis net.Listener
network string
address string
endpoint *url.URL
timeout time.Duration
log *log.Helper
middleware middleware.Middleware
@ -118,19 +119,24 @@ func NewServer(opts ...ServerOption) *Server {
// Endpoint return a real address to registry endpoint.
// examples:
// grpc://127.0.0.1:9000?isSecure=false
func (s *Server) Endpoint() (string, error) {
func (s *Server) Endpoint() (*url.URL, error) {
if s.lis == nil && strings.HasSuffix(s.address, ":0") {
lis, err := net.Listen(s.network, s.address)
if err != nil {
return "", err
return nil, err
}
s.lis = lis
}
addr, err := host.Extract(s.address, s.lis)
if err != nil {
return "", err
return nil, err
}
return fmt.Sprintf("grpc://%s", addr), nil
u := &url.URL{
Scheme: "grpc",
Host: addr,
}
s.endpoint = u
return u, nil
}
// Start start the gRPC server.
@ -161,7 +167,7 @@ func (s *Server) unaryServerInterceptor() grpc.UnaryServerInterceptor {
ctx, cancel := ic.Merge(ctx, s.ctx)
defer cancel()
ctx = transport.NewContext(ctx, transport.Transport{Kind: transport.KindGRPC})
ctx = NewServerContext(ctx, ServerInfo{Server: info.Server, FullMethod: info.FullMethod})
ctx = NewServerContext(ctx, ServerInfo{Server: info.Server, FullMethod: info.FullMethod, Endpoint: s.endpoint})
if s.timeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, s.timeout)

2
transport/grpc/server_test.go
View File

@ -15,7 +15,7 @@ func TestServer(t *testing.T) {
ctx := context.Background()
ctx = context.WithValue(ctx, testKey{}, "test")
srv := NewServer()
if e, err := srv.Endpoint(); err != nil || e == "" {
if e, err := srv.Endpoint(); err != nil || e == nil {
t.Fatal(e, err)
}

2
transport/http/context.go
View File

@ -3,12 +3,14 @@ package http
import (
"context"
"net/http"
"net/url"
)
// ServerInfo represent HTTP server information.
type ServerInfo struct {
Request *http.Request
Response http.ResponseWriter
Endpoint *url.URL
}
type serverKey struct{}

32
transport/http/server.go
View File

@ -3,9 +3,9 @@ package http
import (
"context"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"strings"
"time"
@ -54,13 +54,14 @@ func Logger(logger log.Logger) ServerOption {
// Server is an HTTP server wrapper.
type Server struct {
*http.Server
ctx context.Context
lis net.Listener
network string
address string
timeout time.Duration
router *mux.Router
log *log.Helper
ctx context.Context
lis net.Listener
network string
address string
endpoint *url.URL
timeout time.Duration
router *mux.Router
log *log.Helper
}
// NewServer creates an HTTP server by options.
@ -99,7 +100,7 @@ func (s *Server) ServeHTTP(res http.ResponseWriter, req *http.Request) {
ctx, cancel := ic.Merge(req.Context(), s.ctx)
defer cancel()
ctx = transport.NewContext(ctx, transport.Transport{Kind: transport.KindHTTP})
ctx = NewServerContext(ctx, ServerInfo{Request: req, Response: res})
ctx = NewServerContext(ctx, ServerInfo{Request: req, Response: res, Endpoint: s.endpoint})
if s.timeout > 0 {
ctx, cancel = context.WithTimeout(ctx, s.timeout)
defer cancel()
@ -110,19 +111,24 @@ func (s *Server) ServeHTTP(res http.ResponseWriter, req *http.Request) {
// Endpoint return a real address to registry endpoint.
// examples:
// http://127.0.0.1:8000?isSecure=false
func (s *Server) Endpoint() (string, error) {
func (s *Server) Endpoint() (*url.URL, error) {
if s.lis == nil && strings.HasSuffix(s.address, ":0") {
lis, err := net.Listen(s.network, s.address)
if err != nil {
return "", err
return nil, err
}
s.lis = lis
}
addr, err := host.Extract(s.address, s.lis)
if err != nil {
return "", err
return nil, err
}
return fmt.Sprintf("http://%s", addr), nil
u := &url.URL{
Scheme: "http",
Host: addr,
}
s.endpoint = u
return u, nil
}
// Start start the HTTP server.

2
transport/http/server_test.go
View File

@ -32,7 +32,7 @@ func TestServer(t *testing.T) {
srv := NewServer()
srv.HandleFunc("/index", fn)
if e, err := srv.Endpoint(); err != nil || e == "" {
if e, err := srv.Endpoint(); err != nil || e == nil {
t.Fatal(e, err)
}

3
transport/transport.go
View File

@ -2,6 +2,7 @@ package transport
import (
"context"
"net/url"
// init encoding
_ "github.com/go-kratos/kratos/v2/encoding/json"
@ -18,7 +19,7 @@ type Server interface {
// Endpointer is registry endpoint.
type Endpointer interface {
Endpoint() (string, error)
Endpoint() (*url.URL, error)
}
// Transport is transport context value.