micro/server/rpc_server.go

389 lines
7.8 KiB
Go
Raw Normal View History

2015-01-14 02:31:27 +03:00
package server
import (
"fmt"
2016-03-14 14:01:10 +03:00
"log"
2015-12-21 20:18:04 +03:00
"runtime/debug"
"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"
2015-11-20 19:17:33 +03:00
"github.com/micro/go-micro/registry"
"github.com/micro/go-micro/transport"
2015-05-23 13:53:40 +03:00
2015-05-21 00:57:19 +03:00
"golang.org/x/net/context"
2015-01-14 02:31:27 +03:00
)
2015-05-23 19:40:53 +03:00
type rpcServer struct {
rpc *server
2015-05-27 00:39:48 +03:00
exit chan chan error
sync.RWMutex
opts Options
handlers map[string]Handler
subscribers map[*subscriber][]broker.Subscriber
// used for first registration
registered bool
2015-01-14 02:31:27 +03:00
}
2015-05-27 00:39:48 +03:00
func newRpcServer(opts ...Option) Server {
options := newOptions(opts...)
2015-05-23 19:40:53 +03:00
return &rpcServer{
opts: options,
rpc: &server{
name: options.Name,
serviceMap: make(map[string]*service),
hdlrWrappers: options.HdlrWrappers,
},
handlers: make(map[string]Handler),
subscribers: make(map[*subscriber][]broker.Subscriber),
exit: make(chan chan error),
2015-05-23 19:40:53 +03:00
}
}
2015-01-14 02:31:27 +03:00
2015-05-23 19:40:53 +03:00
func (s *rpcServer) accept(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 {
2016-03-14 14:01:10 +03:00
log.Print(r, string(debug.Stack()))
2015-12-21 20:18:04 +03:00
}
}()
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
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"]
cf, err := s.newCodec(ct)
// TODO: needs better error handling
if err != nil {
sock.Send(&transport.Message{
Header: map[string]string{
"Content-Type": "text/plain",
},
Body: []byte(err.Error()),
})
return
}
2015-11-25 22:50:05 +03:00
2016-05-13 17:58:53 +03:00
codec := newRpcPlusCodec(&msg, sock, cf)
2015-01-14 02:31:27 +03:00
2016-05-13 17:58:53 +03:00
// strip our headers
hdr := make(map[string]string)
for k, v := range msg.Header {
hdr[k] = v
}
delete(hdr, "Content-Type")
delete(hdr, "Timeout")
2015-01-14 02:31:27 +03:00
2016-05-13 17:58:53 +03:00
ctx := metadata.NewContext(context.Background(), hdr)
2015-12-02 23:56:50 +03:00
2016-05-13 17:58:53 +03:00
// set the timeout if we have it
if len(to) > 0 {
if n, err := strconv.ParseUint(to, 10, 64); err == nil {
ctx, _ = context.WithTimeout(ctx, time.Duration(n))
}
2016-05-13 01:32:58 +03:00
}
2016-05-13 17:58:53 +03:00
// TODO: needs better error handling
if err := s.rpc.serveRequest(ctx, codec, ct); err != nil {
log.Printf("Unexpected error serving request, closing socket: %v", err)
return
}
}
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 {
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 internal server
s.rpc = &server{
name: s.opts.Name,
serviceMap: s.rpc.serviceMap,
hdlrWrappers: s.opts.HdlrWrappers,
}
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 {
return newRpcHandler(h, opts...)
2015-01-14 02:31:27 +03:00
}
func (s *rpcServer) Handle(h Handler) error {
s.Lock()
defer s.Unlock()
2015-12-02 04:18:32 +03:00
if err := s.rpc.register(h.Handler()); 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()
_, ok = s.subscribers[sub]
if ok {
return fmt.Errorf("subscriber %v already exists", s)
}
s.subscribers[sub] = nil
s.Unlock()
return nil
}
2016-01-27 15:23:18 +03:00
func (s *rpcServer) Register() error {
// parse address for host, port
config := s.Options()
2015-11-11 21:22:04 +03:00
var advt, host string
var port int
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
}
parts := strings.Split(advt, ":")
if len(parts) > 1 {
host = strings.Join(parts[:len(parts)-1], ":")
port, _ = strconv.Atoi(parts[len(parts)-1])
} else {
host = parts[0]
2015-01-14 02:31:27 +03:00
}
addr, err := extractAddress(host)
if err != nil {
return err
}
// register service
node := &registry.Node{
2016-01-02 03:38:57 +03:00
Id: config.Name + "-" + config.Id,
Address: addr,
Port: port,
Metadata: config.Metadata,
}
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()
2015-12-20 00:56:14 +03:00
s.RLock()
var endpoints []*registry.Endpoint
for _, e := range s.handlers {
// Only advertise non internal handlers
if !e.Options().Internal {
endpoints = append(endpoints, e.Endpoints()...)
}
}
for e, _ := range s.subscribers {
// Only advertise non internal subscribers
if !e.Options().Internal {
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 {
2016-03-14 14:01:10 +03:00
log.Printf("Registering node: %s", 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
}
sub, err := config.Broker.Subscribe(sb.Topic(), handler, opts...)
if err != nil {
return err
}
s.subscribers[sb] = []broker.Subscriber{sub}
}
return nil
}
func (s *rpcServer) Deregister() error {
config := s.Options()
2015-11-11 21:22:04 +03:00
var advt, host string
var port int
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
}
parts := strings.Split(advt, ":")
if len(parts) > 1 {
host = strings.Join(parts[:len(parts)-1], ":")
port, _ = strconv.Atoi(parts[len(parts)-1])
} else {
host = parts[0]
}
addr, err := extractAddress(host)
if err != nil {
return err
}
node := &registry.Node{
2016-01-02 03:38:57 +03:00
Id: config.Name + "-" + config.Id,
Address: addr,
Port: port,
}
service := &registry.Service{
Name: config.Name,
Version: config.Version,
Nodes: []*registry.Node{node},
}
2016-03-14 14:01:10 +03:00
log.Printf("Deregistering node: %s", 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 {
2016-03-14 14:01:10 +03:00
log.Printf("Unsubscribing from topic: %s", 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 {
registerDebugHandler(s)
config := s.Options()
2015-01-14 02:31:27 +03:00
ts, err := config.Transport.Listen(config.Address)
2015-01-14 02:31:27 +03:00
if err != nil {
return err
}
2016-03-14 14:01:10 +03:00
log.Printf("Listening on %s", ts.Addr())
s.Lock()
s.opts.Address = ts.Addr()
s.Unlock()
2015-05-21 23:08:19 +03:00
go ts.Accept(s.accept)
2015-01-14 02:31:27 +03:00
go func() {
ch := <-s.exit
2015-05-21 00:57:19 +03:00
ch <- ts.Close()
config.Broker.Disconnect()
2015-01-14 02:31:27 +03:00
}()
// TODO: subscribe to cruft
return config.Broker.Connect()
2015-01-14 02:31:27 +03:00
}
2015-05-23 19:40:53 +03:00
func (s *rpcServer) Stop() error {
2015-01-14 02:31:27 +03:00
ch := make(chan error)
s.exit <- ch
return <-ch
}
2015-12-20 00:56:14 +03:00
func (s *rpcServer) String() string {
return "rpc"
}