2019-07-26 00:37:51 +02:00
|
|
|
package service
|
|
|
|
|
|
|
|
import (
|
2019-07-26 13:43:17 +02:00
|
|
|
"context"
|
2019-07-29 19:57:40 +02:00
|
|
|
"errors"
|
2019-07-26 18:11:59 +02:00
|
|
|
"fmt"
|
|
|
|
"io"
|
2019-07-26 00:37:51 +02:00
|
|
|
"sync"
|
2019-07-26 18:11:59 +02:00
|
|
|
"time"
|
2019-07-26 00:37:51 +02:00
|
|
|
|
|
|
|
"github.com/micro/go-micro/client"
|
2019-08-05 18:44:33 +02:00
|
|
|
"github.com/micro/go-micro/router"
|
|
|
|
pb "github.com/micro/go-micro/router/proto"
|
2019-07-26 00:37:51 +02:00
|
|
|
)
|
|
|
|
|
|
|
|
type svc struct {
|
2019-07-29 13:52:32 +02:00
|
|
|
sync.RWMutex
|
2019-07-26 18:11:59 +02:00
|
|
|
opts router.Options
|
2019-07-29 19:57:40 +02:00
|
|
|
callOpts []client.CallOption
|
2019-07-26 18:11:59 +02:00
|
|
|
router pb.RouterService
|
2019-07-29 19:57:40 +02:00
|
|
|
table *table
|
|
|
|
status *router.Status
|
|
|
|
exit chan bool
|
2019-07-26 18:11:59 +02:00
|
|
|
errChan chan error
|
|
|
|
advertChan chan *router.Advert
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// NewRouter creates new service router and returns it
|
|
|
|
func NewRouter(opts ...router.Option) router.Router {
|
|
|
|
// get default options
|
|
|
|
options := router.DefaultOptions()
|
|
|
|
|
|
|
|
// apply requested options
|
|
|
|
for _, o := range opts {
|
|
|
|
o(&options)
|
|
|
|
}
|
|
|
|
|
|
|
|
// NOTE: might need some client opts here
|
2019-07-29 19:57:40 +02:00
|
|
|
cli := client.DefaultClient
|
|
|
|
|
|
|
|
// set options client
|
|
|
|
if options.Client != nil {
|
|
|
|
cli = options.Client
|
|
|
|
}
|
2019-07-26 00:37:51 +02:00
|
|
|
|
2019-08-12 19:18:17 +02:00
|
|
|
// set the status to Stopped
|
|
|
|
status := &router.Status{
|
|
|
|
Code: router.Stopped,
|
|
|
|
Error: nil,
|
|
|
|
}
|
|
|
|
|
2019-07-26 00:37:51 +02:00
|
|
|
// NOTE: should we have Client/Service option in router.Options?
|
|
|
|
s := &svc{
|
2019-07-29 19:57:40 +02:00
|
|
|
opts: options,
|
2019-08-12 19:18:17 +02:00
|
|
|
status: status,
|
2019-07-29 19:57:40 +02:00
|
|
|
router: pb.NewRouterService(router.DefaultName, cli),
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
// set the router address to call
|
|
|
|
if len(options.Address) > 0 {
|
|
|
|
s.callOpts = []client.CallOption{
|
|
|
|
client.WithAddress(options.Address),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
// set the table
|
|
|
|
s.table = &table{pb.NewTableService(router.DefaultName, cli), s.callOpts}
|
2019-07-26 18:11:59 +02:00
|
|
|
|
2019-07-26 00:37:51 +02:00
|
|
|
return s
|
|
|
|
}
|
|
|
|
|
|
|
|
// Init initializes router with given options
|
|
|
|
func (s *svc) Init(opts ...router.Option) error {
|
2019-08-12 19:18:17 +02:00
|
|
|
s.Lock()
|
|
|
|
defer s.Unlock()
|
|
|
|
|
2019-07-26 00:37:51 +02:00
|
|
|
for _, o := range opts {
|
|
|
|
o(&s.opts)
|
|
|
|
}
|
2019-08-12 19:18:17 +02:00
|
|
|
|
2019-07-26 00:37:51 +02:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Options returns router options
|
|
|
|
func (s *svc) Options() router.Options {
|
2019-08-12 19:18:17 +02:00
|
|
|
s.Lock()
|
|
|
|
opts := s.opts
|
|
|
|
s.Unlock()
|
|
|
|
|
|
|
|
return opts
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
2019-08-12 19:18:17 +02:00
|
|
|
// Table returns routing table
|
2019-07-29 19:57:40 +02:00
|
|
|
func (s *svc) Table() router.Table {
|
|
|
|
return s.table
|
2019-07-26 18:11:59 +02:00
|
|
|
}
|
|
|
|
|
2019-08-12 19:18:17 +02:00
|
|
|
// Start starts the service
|
|
|
|
func (s *svc) Start() error {
|
|
|
|
s.Lock()
|
|
|
|
defer s.Unlock()
|
|
|
|
|
|
|
|
s.status = &router.Status{
|
|
|
|
Code: router.Running,
|
|
|
|
Error: nil,
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
func (s *svc) advertiseEvents(advertChan chan *router.Advert, stream pb.Router_AdvertiseService) error {
|
2019-07-27 17:00:55 +02:00
|
|
|
go func() {
|
|
|
|
<-s.exit
|
|
|
|
stream.Close()
|
|
|
|
}()
|
|
|
|
|
2019-07-26 18:11:59 +02:00
|
|
|
var advErr error
|
|
|
|
|
|
|
|
for {
|
|
|
|
resp, err := stream.Recv()
|
|
|
|
if err != nil {
|
|
|
|
if err != io.EOF {
|
|
|
|
advErr = err
|
|
|
|
}
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
2019-07-27 13:40:16 +02:00
|
|
|
events := make([]*router.Event, len(resp.Events))
|
|
|
|
for i, event := range resp.Events {
|
|
|
|
route := router.Route{
|
|
|
|
Service: event.Route.Service,
|
|
|
|
Address: event.Route.Address,
|
|
|
|
Gateway: event.Route.Gateway,
|
|
|
|
Network: event.Route.Network,
|
|
|
|
Link: event.Route.Link,
|
|
|
|
Metric: int(event.Route.Metric),
|
|
|
|
}
|
|
|
|
|
|
|
|
events[i] = &router.Event{
|
|
|
|
Type: router.EventType(event.Type),
|
|
|
|
Timestamp: time.Unix(0, event.Timestamp),
|
|
|
|
Route: route,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-07-26 18:11:59 +02:00
|
|
|
advert := &router.Advert{
|
|
|
|
Id: resp.Id,
|
|
|
|
Type: router.AdvertType(resp.Type),
|
|
|
|
Timestamp: time.Unix(0, resp.Timestamp),
|
2019-07-27 13:40:16 +02:00
|
|
|
TTL: time.Duration(resp.Ttl),
|
|
|
|
Events: events,
|
2019-07-26 18:11:59 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
select {
|
2019-07-29 19:57:40 +02:00
|
|
|
case advertChan <- advert:
|
2019-07-26 18:11:59 +02:00
|
|
|
case <-s.exit:
|
2019-07-29 19:57:40 +02:00
|
|
|
close(advertChan)
|
2019-07-26 18:11:59 +02:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-07-27 17:00:55 +02:00
|
|
|
// close the channel on exit
|
2019-07-29 19:57:40 +02:00
|
|
|
close(advertChan)
|
2019-07-27 17:00:55 +02:00
|
|
|
|
2019-07-26 18:11:59 +02:00
|
|
|
return advErr
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
|
|
|
|
2019-07-26 00:37:51 +02:00
|
|
|
// Advertise advertises routes to the network
|
|
|
|
func (s *svc) Advertise() (<-chan *router.Advert, error) {
|
2019-07-26 18:11:59 +02:00
|
|
|
s.Lock()
|
|
|
|
defer s.Unlock()
|
|
|
|
|
2019-08-12 19:18:17 +02:00
|
|
|
switch s.status.Code {
|
2019-07-29 19:57:40 +02:00
|
|
|
case router.Running, router.Advertising:
|
2019-08-05 18:47:50 +02:00
|
|
|
stream, err := s.router.Advertise(context.Background(), &pb.Request{}, s.callOpts...)
|
2019-07-26 18:11:59 +02:00
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("failed getting advert stream: %s", err)
|
|
|
|
}
|
|
|
|
// create advertise and event channels
|
2019-07-29 19:57:40 +02:00
|
|
|
advertChan := make(chan *router.Advert)
|
|
|
|
go s.advertiseEvents(advertChan, stream)
|
|
|
|
return advertChan, nil
|
2019-07-26 18:11:59 +02:00
|
|
|
case router.Stopped:
|
2019-08-12 19:18:17 +02:00
|
|
|
return nil, fmt.Errorf("not running")
|
2019-07-26 18:11:59 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil, fmt.Errorf("error: %s", s.status.Error)
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Process processes incoming adverts
|
2019-07-27 13:40:16 +02:00
|
|
|
func (s *svc) Process(advert *router.Advert) error {
|
|
|
|
var events []*pb.Event
|
|
|
|
for _, event := range advert.Events {
|
|
|
|
route := &pb.Route{
|
|
|
|
Service: event.Route.Service,
|
|
|
|
Address: event.Route.Address,
|
|
|
|
Gateway: event.Route.Gateway,
|
|
|
|
Network: event.Route.Network,
|
|
|
|
Link: event.Route.Link,
|
|
|
|
Metric: int64(event.Route.Metric),
|
|
|
|
}
|
|
|
|
e := &pb.Event{
|
|
|
|
Type: pb.EventType(event.Type),
|
|
|
|
Timestamp: event.Timestamp.UnixNano(),
|
|
|
|
Route: route,
|
|
|
|
}
|
|
|
|
events = append(events, e)
|
|
|
|
}
|
|
|
|
|
|
|
|
advertReq := &pb.Advert{
|
|
|
|
Id: s.Options().Id,
|
|
|
|
Type: pb.AdvertType(advert.Type),
|
|
|
|
Timestamp: advert.Timestamp.UnixNano(),
|
|
|
|
Events: events,
|
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
if _, err := s.router.Process(context.Background(), advertReq, s.callOpts...); err != nil {
|
2019-07-27 13:40:16 +02:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2019-07-26 00:37:51 +02:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
// Status returns router status
|
|
|
|
func (s *svc) Status() router.Status {
|
|
|
|
s.Lock()
|
|
|
|
defer s.Unlock()
|
2019-07-26 15:05:03 +02:00
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
// check if its stopped
|
|
|
|
select {
|
|
|
|
case <-s.exit:
|
|
|
|
return router.Status{
|
|
|
|
Code: router.Stopped,
|
|
|
|
Error: nil,
|
|
|
|
}
|
|
|
|
default:
|
|
|
|
// don't block
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
// check the remote router
|
|
|
|
rsp, err := s.router.Status(context.Background(), &pb.Request{}, s.callOpts...)
|
|
|
|
if err != nil {
|
|
|
|
return router.Status{
|
|
|
|
Code: router.Error,
|
|
|
|
Error: err,
|
|
|
|
}
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
code := router.Running
|
|
|
|
var serr error
|
|
|
|
|
|
|
|
switch rsp.Status.Code {
|
|
|
|
case "running":
|
|
|
|
code = router.Running
|
|
|
|
case "advertising":
|
|
|
|
code = router.Advertising
|
|
|
|
case "stopped":
|
|
|
|
code = router.Stopped
|
|
|
|
case "error":
|
|
|
|
code = router.Error
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
if len(rsp.Status.Error) > 0 {
|
|
|
|
serr = errors.New(rsp.Status.Error)
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
return router.Status{
|
|
|
|
Code: code,
|
|
|
|
Error: serr,
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
// Remote router cannot be stopped
|
|
|
|
func (s *svc) Stop() error {
|
|
|
|
s.Lock()
|
|
|
|
defer s.Unlock()
|
2019-07-26 13:43:17 +02:00
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
select {
|
|
|
|
case <-s.exit:
|
|
|
|
return nil
|
|
|
|
default:
|
|
|
|
close(s.exit)
|
2019-07-26 13:43:17 +02:00
|
|
|
}
|
|
|
|
|
2019-07-29 19:57:40 +02:00
|
|
|
return nil
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Lookup looks up routes in the routing table and returns them
|
|
|
|
func (s *svc) Lookup(q router.Query) ([]router.Route, error) {
|
2019-07-26 13:43:17 +02:00
|
|
|
// call the router
|
|
|
|
resp, err := s.router.Lookup(context.Background(), &pb.LookupRequest{
|
|
|
|
Query: &pb.Query{
|
|
|
|
Service: q.Options().Service,
|
|
|
|
Gateway: q.Options().Gateway,
|
|
|
|
Network: q.Options().Network,
|
|
|
|
},
|
2019-07-29 19:57:40 +02:00
|
|
|
}, s.callOpts...)
|
2019-07-26 13:43:17 +02:00
|
|
|
|
|
|
|
// errored out
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
routes := make([]router.Route, len(resp.Routes))
|
|
|
|
for i, route := range resp.Routes {
|
|
|
|
routes[i] = router.Route{
|
|
|
|
Service: route.Service,
|
|
|
|
Address: route.Address,
|
|
|
|
Gateway: route.Gateway,
|
|
|
|
Network: route.Network,
|
|
|
|
Link: route.Link,
|
|
|
|
Metric: int(route.Metric),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return routes, nil
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Watch returns a watcher which allows to track updates to the routing table
|
|
|
|
func (s *svc) Watch(opts ...router.WatchOption) (router.Watcher, error) {
|
2019-07-29 19:57:40 +02:00
|
|
|
rsp, err := s.router.Watch(context.Background(), &pb.WatchRequest{}, s.callOpts...)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2019-07-26 15:05:03 +02:00
|
|
|
}
|
2019-07-29 19:57:40 +02:00
|
|
|
var options router.WatchOptions
|
2019-07-26 15:05:03 +02:00
|
|
|
for _, o := range opts {
|
2019-07-29 19:57:40 +02:00
|
|
|
o(&options)
|
2019-07-26 18:11:59 +02:00
|
|
|
}
|
2019-07-29 19:57:40 +02:00
|
|
|
return newWatcher(rsp, options)
|
2019-07-26 00:37:51 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Returns the router implementation
|
|
|
|
func (s *svc) String() string {
|
|
|
|
return "service"
|
|
|
|
}
|