micro/server/rpc_server.go

753 lines
16 KiB
Go
Raw Normal View History

2015-01-14 02:31:27 +03:00
package server
import (
2018-03-03 14:53:52 +03:00
"context"
"fmt"
"net"
2015-12-21 20:18:04 +03:00
"runtime/debug"
"sort"
"strconv"
"strings"
"sync"
2016-05-13 01:32:58 +03:00
"time"
2015-11-20 19:17:33 +03:00
"github.com/micro/go-micro/broker"
"github.com/micro/go-micro/codec"
2016-01-28 20:55:28 +03:00
"github.com/micro/go-micro/metadata"
2019-07-07 14:33:54 +03:00
"github.com/micro/go-micro/registry"
2019-07-10 09:47:17 +03:00
"github.com/micro/go-micro/transport"
2019-05-31 01:52:10 +03:00
"github.com/micro/go-micro/util/addr"
2019-05-31 02:38:05 +03:00
log "github.com/micro/go-micro/util/log"
mnet "github.com/micro/go-micro/util/net"
2019-08-15 22:08:49 +03:00
"github.com/micro/go-micro/util/socket"
2015-01-14 02:31:27 +03:00
)
2015-05-23 19:40:53 +03:00
type rpcServer struct {
2019-01-07 17:44:40 +03:00
router *router
exit chan chan error
sync.RWMutex
opts Options
handlers map[string]Handler
subscribers map[*subscriber][]broker.Subscriber
// marks the serve as started
started bool
// used for first registration
registered bool
// graceful exit
wg *sync.WaitGroup
2015-01-14 02:31:27 +03:00
}
2015-05-27 00:39:48 +03:00
func newRpcServer(opts ...Option) Server {
options := newOptions(opts...)
router := newRpcRouter()
router.hdlrWrappers = options.HdlrWrappers
2015-05-23 19:40:53 +03:00
return &rpcServer{
2019-01-07 17:44:40 +03:00
opts: options,
2019-06-12 14:05:34 +03:00
router: router,
handlers: make(map[string]Handler),
subscribers: make(map[*subscriber][]broker.Subscriber),
exit: make(chan chan error),
wg: wait(options.Context),
2015-05-23 19:40:53 +03:00
}
}
2015-01-14 02:31:27 +03:00
type rpcRouter struct {
h func(context.Context, Request, interface{}) error
}
func (r rpcRouter) ServeRequest(ctx context.Context, req Request, rsp Response) error {
return r.h(ctx, req, rsp)
}
2019-01-08 23:32:47 +03:00
// ServeConn serves a single connection
func (s *rpcServer) ServeConn(sock transport.Socket) {
2015-12-21 20:18:04 +03:00
defer func() {
2016-05-10 12:55:18 +03:00
// close socket
sock.Close()
2015-12-21 20:18:04 +03:00
if r := recover(); r != nil {
2017-05-16 21:14:00 +03:00
log.Log("panic recovered: ", r)
log.Log(string(debug.Stack()))
2015-12-21 20:18:04 +03:00
}
}()
2019-08-15 22:08:49 +03:00
// multiplex the streams on a single socket by Micro-Stream
var mtx sync.RWMutex
sockets := make(map[string]*socket.Socket)
2016-05-13 17:58:53 +03:00
for {
var msg transport.Message
if err := sock.Recv(&msg); err != nil {
return
}
2015-01-14 02:31:27 +03:00
2019-08-16 16:42:45 +03:00
// use Micro-Stream as the stream identifier
2019-08-15 22:08:49 +03:00
// in the event its blank we'll always process
// on the same socket
id := msg.Header["Micro-Stream"]
// if there's no stream id then its a standard request
2019-08-16 16:42:45 +03:00
// use the Micro-Id
2019-08-15 22:08:49 +03:00
if len(id) == 0 {
id = msg.Header["Micro-Id"]
}
// add to wait group if "wait" is opt-in
if s.wg != nil {
s.wg.Add(1)
}
2018-05-26 11:41:41 +03:00
2019-08-15 22:08:49 +03:00
// check we have an existing socket
mtx.RLock()
psock, ok := sockets[id]
mtx.RUnlock()
// got the socket
if ok {
// accept the message
if err := psock.Accept(&msg); err != nil {
// delete the socket
mtx.Lock()
delete(sockets, id)
mtx.Unlock()
}
// done(1)
if s.wg != nil {
s.wg.Done()
}
// continue to the next message
continue
}
// no socket was found
psock = socket.New()
psock.SetLocal(sock.Local())
psock.SetRemote(sock.Remote())
// load the socket
psock.Accept(&msg)
// save a new socket
mtx.Lock()
sockets[id] = psock
mtx.Unlock()
// process the outbound messages from the socket
go func(id string, psock *socket.Socket) {
defer psock.Close()
for {
// get the message from our internal handler/stream
m := new(transport.Message)
if err := psock.Process(m); err != nil {
// delete the socket
mtx.Lock()
delete(sockets, id)
mtx.Unlock()
return
}
// send the message back over the socket
if err := sock.Send(m); err != nil {
return
}
}
}(id, psock)
// now walk the usual path
2016-05-13 17:58:53 +03:00
// we use this Timeout header to set a server deadline
to := msg.Header["Timeout"]
// we use this Content-Type header to identify the codec needed
ct := msg.Header["Content-Type"]
2019-08-15 22:08:49 +03:00
// copy the message headers
2019-01-07 21:20:47 +03:00
hdr := make(map[string]string)
for k, v := range msg.Header {
hdr[k] = v
}
// set local/remote ips
hdr["Local"] = sock.Local()
hdr["Remote"] = sock.Remote()
2019-08-15 22:08:49 +03:00
// create new context with the metadata
2019-01-07 21:20:47 +03:00
ctx := metadata.NewContext(context.Background(), hdr)
2019-08-15 22:08:49 +03:00
// set the timeout from the header if we have it
2019-01-07 21:20:47 +03:00
if len(to) > 0 {
if n, err := strconv.ParseUint(to, 10, 64); err == nil {
ctx, _ = context.WithTimeout(ctx, time.Duration(n))
}
}
2019-08-15 22:08:49 +03:00
// if there's no content type default it
if len(ct) == 0 {
2019-01-08 23:32:47 +03:00
msg.Header["Content-Type"] = DefaultContentType
ct = DefaultContentType
}
2019-01-18 13:12:57 +03:00
// setup old protocol
cf := setupProtocol(&msg)
// no old codec
if cf == nil {
// TODO: needs better error handling
var err error
if cf, err = s.newCodec(ct); err != nil {
sock.Send(&transport.Message{
Header: map[string]string{
"Content-Type": "text/plain",
},
Body: []byte(err.Error()),
})
if s.wg != nil {
s.wg.Done()
}
2019-01-18 13:12:57 +03:00
return
}
2016-05-13 17:58:53 +03:00
}
2015-11-25 22:50:05 +03:00
2019-08-15 22:08:49 +03:00
rcodec := newRpcCodec(&msg, psock, cf)
// check stream id
var stream bool
if v := getHeader("Micro-Stream", msg.Header); len(v) > 0 {
stream = true
}
2019-01-09 22:28:13 +03:00
2019-01-09 19:20:57 +03:00
// internal request
request := &rpcRequest{
2019-01-24 13:11:02 +03:00
service: getHeader("Micro-Service", msg.Header),
method: getHeader("Micro-Method", msg.Header),
endpoint: getHeader("Micro-Endpoint", msg.Header),
2019-01-09 19:20:57 +03:00
contentType: ct,
2019-01-13 15:15:35 +03:00
codec: rcodec,
2019-01-09 22:11:47 +03:00
header: msg.Header,
2019-01-09 19:20:57 +03:00
body: msg.Body,
2019-08-15 22:08:49 +03:00
socket: psock,
stream: stream,
2019-01-09 19:20:57 +03:00
}
2015-01-14 02:31:27 +03:00
2019-01-09 22:11:47 +03:00
// internal response
response := &rpcResponse{
header: make(map[string]string),
2019-08-15 22:08:49 +03:00
socket: psock,
2019-01-13 15:15:35 +03:00
codec: rcodec,
2019-01-09 22:11:47 +03:00
}
2019-01-08 23:32:47 +03:00
2019-01-09 22:11:47 +03:00
// set router
r := Router(s.router)
2019-01-11 18:49:54 +03:00
// if not nil use the router specified
if s.opts.Router != nil {
// create a wrapped function
handler := func(ctx context.Context, req Request, rsp interface{}) error {
return s.opts.Router.ServeRequest(ctx, req, rsp.(Response))
}
2019-01-08 23:32:47 +03:00
// execute the wrapper for it
for i := len(s.opts.HdlrWrappers); i > 0; i-- {
handler = s.opts.HdlrWrappers[i-1](handler)
}
2019-01-11 18:49:54 +03:00
// set the router
r = rpcRouter{handler}
2019-01-11 18:49:54 +03:00
}
2019-08-15 22:08:49 +03:00
// serve the request in a go routine as this may be a stream
go func(id string, psock *socket.Socket) {
// serve the actual request using the request router
if err := r.ServeRequest(ctx, request, response); err != nil {
// write an error response
err = rcodec.Write(&codec.Message{
Header: msg.Header,
Error: err.Error(),
Type: codec.Error,
}, nil)
// could not write the error response
if err != nil {
log.Logf("rpc: unable to write error response: %v", err)
}
2019-01-22 16:55:04 +03:00
}
2019-08-15 22:08:49 +03:00
mtx.Lock()
delete(sockets, id)
mtx.Unlock()
// signal we're done
if s.wg != nil {
s.wg.Done()
}
2019-08-15 22:08:49 +03:00
}(id, psock)
2019-01-11 18:49:54 +03:00
}
2015-05-21 00:57:19 +03:00
}
func (s *rpcServer) newCodec(contentType string) (codec.NewCodec, error) {
if cf, ok := s.opts.Codecs[contentType]; ok {
return cf, nil
2015-11-25 22:50:05 +03:00
}
if cf, ok := DefaultCodecs[contentType]; ok {
2015-11-25 22:50:05 +03:00
return cf, nil
}
return nil, fmt.Errorf("Unsupported Content-Type: %s", contentType)
}
func (s *rpcServer) Options() Options {
s.RLock()
opts := s.opts
s.RUnlock()
return opts
2015-01-14 02:31:27 +03:00
}
2016-01-02 22:12:17 +03:00
func (s *rpcServer) Init(opts ...Option) error {
s.Lock()
2015-05-27 00:39:48 +03:00
for _, opt := range opts {
opt(&s.opts)
}
// update router if its the default
if s.opts.Router == nil {
r := newRpcRouter()
r.hdlrWrappers = s.opts.HdlrWrappers
r.serviceMap = s.router.serviceMap
s.router = r
}
s.Unlock()
2016-01-02 22:12:17 +03:00
return nil
2015-01-14 02:31:27 +03:00
}
func (s *rpcServer) NewHandler(h interface{}, opts ...HandlerOption) Handler {
2019-01-07 20:17:06 +03:00
return s.router.NewHandler(h, opts...)
2015-01-14 02:31:27 +03:00
}
func (s *rpcServer) Handle(h Handler) error {
s.Lock()
defer s.Unlock()
2019-01-07 20:17:06 +03:00
if err := s.router.Handle(h); err != nil {
return err
}
s.handlers[h.Name()] = h
return nil
2015-01-14 02:31:27 +03:00
}
func (s *rpcServer) NewSubscriber(topic string, sb interface{}, opts ...SubscriberOption) Subscriber {
return newSubscriber(topic, sb, opts...)
}
func (s *rpcServer) Subscribe(sb Subscriber) error {
sub, ok := sb.(*subscriber)
if !ok {
return fmt.Errorf("invalid subscriber: expected *subscriber")
}
if len(sub.handlers) == 0 {
return fmt.Errorf("invalid subscriber: no handler functions")
}
if err := validateSubscriber(sb); err != nil {
return err
}
s.Lock()
defer s.Unlock()
_, ok = s.subscribers[sub]
if ok {
return fmt.Errorf("subscriber %v already exists", s)
}
s.subscribers[sub] = nil
return nil
}
2016-01-27 15:23:18 +03:00
func (s *rpcServer) Register() error {
var err error
var advt, host, port string
// parse address for host, port
config := s.Options()
2015-11-11 21:22:04 +03:00
// check the advertise address first
// if it exists then use it, otherwise
// use the address
if len(config.Advertise) > 0 {
advt = config.Advertise
2015-11-11 21:22:04 +03:00
} else {
advt = config.Address
2015-11-11 21:22:04 +03:00
}
if cnt := strings.Count(advt, ":"); cnt >= 1 {
// ipv6 address in format [host]:port or ipv4 host:port
host, port, err = net.SplitHostPort(advt)
if err != nil {
return err
}
} else {
host = advt
2015-01-14 02:31:27 +03:00
}
2017-01-12 16:20:34 +03:00
addr, err := addr.Extract(host)
if err != nil {
return err
}
2019-02-04 13:29:10 +03:00
// make copy of metadata
md := make(metadata.Metadata)
for k, v := range config.Metadata {
md[k] = v
}
2019-08-14 16:32:28 +03:00
// mq-rpc(eg. nats) doesn't need the port. its addr is queue name.
if port != "" {
addr = mnet.HostPort(addr, port)
}
// register service
node := &registry.Node{
2016-01-02 03:38:57 +03:00
Id: config.Name + "-" + config.Id,
2019-08-14 16:32:28 +03:00
Address: addr,
2019-02-04 13:29:10 +03:00
Metadata: md,
}
node.Metadata["transport"] = config.Transport.String()
2016-01-04 04:27:05 +03:00
node.Metadata["broker"] = config.Broker.String()
node.Metadata["server"] = s.String()
2016-01-05 02:27:44 +03:00
node.Metadata["registry"] = config.Registry.String()
2019-01-18 15:30:39 +03:00
node.Metadata["protocol"] = "mucp"
2015-12-20 00:56:14 +03:00
s.RLock()
// Maps are ordered randomly, sort the keys for consistency
var handlerList []string
for n, e := range s.handlers {
// Only advertise non internal handlers
if !e.Options().Internal {
handlerList = append(handlerList, n)
}
}
sort.Strings(handlerList)
var subscriberList []*subscriber
for e := range s.subscribers {
// Only advertise non internal subscribers
if !e.Options().Internal {
subscriberList = append(subscriberList, e)
}
}
sort.Slice(subscriberList, func(i, j int) bool {
return subscriberList[i].topic > subscriberList[j].topic
})
var endpoints []*registry.Endpoint
for _, n := range handlerList {
endpoints = append(endpoints, s.handlers[n].Endpoints()...)
}
for _, e := range subscriberList {
endpoints = append(endpoints, e.Endpoints()...)
}
s.RUnlock()
service := &registry.Service{
Name: config.Name,
Version: config.Version,
Nodes: []*registry.Node{node},
Endpoints: endpoints,
}
2016-02-28 01:14:25 +03:00
s.Lock()
registered := s.registered
s.Unlock()
if !registered {
2019-02-15 20:20:09 +03:00
log.Logf("Registry [%s] Registering node: %s", config.Registry.String(), node.Id)
2016-02-28 01:14:25 +03:00
}
2016-01-27 15:23:18 +03:00
// create registry options
rOpts := []registry.RegisterOption{registry.RegisterTTL(config.RegisterTTL)}
2016-01-27 02:32:27 +03:00
if err := config.Registry.Register(service, rOpts...); err != nil {
return err
}
2016-02-28 01:14:25 +03:00
// already registered? don't need to register subscribers
if registered {
return nil
}
2016-02-28 01:14:25 +03:00
s.Lock()
defer s.Unlock()
s.registered = true
for sb, _ := range s.subscribers {
2015-12-02 22:56:57 +03:00
handler := s.createSubHandler(sb, s.opts)
2016-01-23 00:48:43 +03:00
var opts []broker.SubscribeOption
if queue := sb.Options().Queue; len(queue) > 0 {
2016-05-10 12:55:18 +03:00
opts = append(opts, broker.Queue(queue))
2016-01-23 00:48:43 +03:00
}
2019-05-24 15:06:27 +03:00
if cx := sb.Options().Context; cx != nil {
opts = append(opts, broker.SubscribeContext(cx))
}
2019-05-24 15:06:27 +03:00
if !sb.Options().AutoAck {
opts = append(opts, broker.DisableAutoAck())
}
2016-01-23 00:48:43 +03:00
sub, err := config.Broker.Subscribe(sb.Topic(), handler, opts...)
if err != nil {
return err
}
log.Logf("Subscribing %s to topic: %s", node.Id, sub.Topic())
s.subscribers[sb] = []broker.Subscriber{sub}
}
return nil
}
func (s *rpcServer) Deregister() error {
var err error
var advt, host, port string
config := s.Options()
2015-11-11 21:22:04 +03:00
// check the advertise address first
// if it exists then use it, otherwise
// use the address
if len(config.Advertise) > 0 {
advt = config.Advertise
2015-11-11 21:22:04 +03:00
} else {
advt = config.Address
2015-11-11 21:22:04 +03:00
}
if cnt := strings.Count(advt, ":"); cnt >= 1 {
// ipv6 address in format [host]:port or ipv4 host:port
host, port, err = net.SplitHostPort(advt)
if err != nil {
return err
}
} else {
host = advt
}
2017-01-12 16:20:34 +03:00
addr, err := addr.Extract(host)
if err != nil {
return err
}
2019-08-15 10:59:50 +03:00
// mq-rpc(eg. nats) doesn't need the port. its addr is queue name.
if port != "" {
addr = mnet.HostPort(addr, port)
}
node := &registry.Node{
2016-01-02 03:38:57 +03:00
Id: config.Name + "-" + config.Id,
2019-08-15 10:59:50 +03:00
Address: addr,
}
service := &registry.Service{
Name: config.Name,
Version: config.Version,
Nodes: []*registry.Node{node},
}
2019-02-15 20:20:09 +03:00
log.Logf("Registry [%s] Deregistering node: %s", config.Registry.String(), node.Id)
if err := config.Registry.Deregister(service); err != nil {
return err
}
s.Lock()
if !s.registered {
s.Unlock()
return nil
}
s.registered = false
for sb, subs := range s.subscribers {
for _, sub := range subs {
log.Logf("Unsubscribing %s from topic: %s", node.Id, sub.Topic())
sub.Unsubscribe()
}
s.subscribers[sb] = nil
}
s.Unlock()
return nil
2015-01-14 02:31:27 +03:00
}
2015-05-23 19:40:53 +03:00
func (s *rpcServer) Start() error {
s.RLock()
if s.started {
s.RUnlock()
return nil
}
s.RUnlock()
config := s.Options()
2015-01-14 02:31:27 +03:00
2019-01-24 16:22:17 +03:00
// start listening on the transport
ts, err := config.Transport.Listen(config.Address)
2015-01-14 02:31:27 +03:00
if err != nil {
return err
}
2019-01-22 15:18:33 +03:00
log.Logf("Transport [%s] Listening on %s", config.Transport.String(), ts.Addr())
// swap address
2019-01-24 16:22:17 +03:00
s.Lock()
addr := s.opts.Address
s.opts.Address = ts.Addr()
s.Unlock()
2019-01-24 16:22:17 +03:00
// connect to the broker
if err := config.Broker.Connect(); err != nil {
return err
}
2019-07-07 14:33:47 +03:00
bname := config.Broker.String()
2019-07-10 21:58:30 +03:00
log.Logf("Broker [%s] Connected to %s", bname, config.Broker.Address())
2019-01-24 16:22:17 +03:00
// use RegisterCheck func before register
if err = s.opts.RegisterCheck(s.opts.Context); err != nil {
log.Logf("Server %s-%s register check error: %s", config.Name, config.Id, err)
} else {
// announce self to the world
if err = s.Register(); err != nil {
log.Logf("Server %s-%s register error: %s", config.Name, config.Id, err)
}
2019-01-24 16:22:17 +03:00
}
exit := make(chan bool)
2018-11-15 22:55:13 +03:00
go func() {
for {
2019-01-24 16:22:17 +03:00
// listen for connections
2019-01-08 23:32:47 +03:00
err := ts.Accept(s.ServeConn)
2018-11-15 22:55:13 +03:00
2019-01-24 16:22:17 +03:00
// TODO: listen for messages
// msg := broker.Exchange(service).Consume()
2018-11-15 22:55:13 +03:00
select {
2019-01-24 16:22:17 +03:00
// check if we're supposed to exit
2018-11-15 22:55:13 +03:00
case <-exit:
return
// check the error and backoff
2019-01-24 16:22:17 +03:00
default:
if err != nil {
log.Logf("Accept error: %v", err)
time.Sleep(time.Second)
continue
}
2018-11-15 22:55:13 +03:00
}
// no error just exit
return
}
}()
2015-01-14 02:31:27 +03:00
go func() {
2019-01-24 19:08:04 +03:00
t := new(time.Ticker)
2019-01-24 16:22:17 +03:00
// only process if it exists
2019-01-24 19:08:04 +03:00
if s.opts.RegisterInterval > time.Duration(0) {
// new ticker
t = time.NewTicker(s.opts.RegisterInterval)
2019-01-24 16:22:17 +03:00
}
// return error chan
var ch chan error
Loop:
for {
select {
// register self on interval
case <-t.C:
s.RLock()
registered := s.registered
s.RUnlock()
if err = s.opts.RegisterCheck(s.opts.Context); err != nil && registered {
log.Logf("Server %s-%s register check error: %s, deregister it", config.Name, config.Id, err)
// deregister self in case of error
if err := s.Deregister(); err != nil {
log.Logf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
}
} else {
if err := s.Register(); err != nil {
log.Logf("Server %s-%s register error: %s", config.Name, config.Id, err)
}
2019-01-24 16:22:17 +03:00
}
// wait for exit
case ch = <-s.exit:
t.Stop()
close(exit)
break Loop
}
}
// deregister self
if err := s.Deregister(); err != nil {
log.Logf("Server %s-%s deregister error: %s", config.Name, config.Id, err)
2019-01-24 16:22:17 +03:00
}
// wait for requests to finish
if s.wg != nil {
s.wg.Wait()
}
// close transport listener
2015-05-21 00:57:19 +03:00
ch <- ts.Close()
// disconnect the broker
config.Broker.Disconnect()
// swap back address
2019-01-24 16:22:17 +03:00
s.Lock()
s.opts.Address = addr
s.Unlock()
2015-01-14 02:31:27 +03:00
}()
// mark the server as started
s.Lock()
s.started = true
s.Unlock()
2019-01-22 15:18:33 +03:00
return nil
2015-01-14 02:31:27 +03:00
}
2015-05-23 19:40:53 +03:00
func (s *rpcServer) Stop() error {
s.RLock()
if !s.started {
s.RUnlock()
return nil
}
s.RUnlock()
2015-01-14 02:31:27 +03:00
ch := make(chan error)
s.exit <- ch
var err error
select {
case err = <-ch:
2019-08-23 17:07:08 +03:00
s.Lock()
s.started = false
2019-08-23 17:07:08 +03:00
s.Unlock()
}
return err
2015-01-14 02:31:27 +03:00
}
2015-12-20 00:56:14 +03:00
func (s *rpcServer) String() string {
2019-06-08 21:40:44 +03:00
return "mucp"
2015-12-20 00:56:14 +03:00
}