p2p/discover/mdns/lan_discover.go

   1 package mdns
   2
   3 import (
   4         "net"
   5         "sync/atomic"
   6         "time"
   7
   8         log "github.com/sirupsen/logrus"
   9
  10         "github.com/vapor/event"
  11 )
  12
  13 const (
  14         logModule            = "p2p/mdns"
  15         registerServiceCycle = 10 * time.Minute
  16         registerServiceDelay = 2 * time.Second
  17 )
  18
  19 // LANPeerEvent represent LAN peer ip and port.
  20 type LANPeerEvent struct {
  21         IP   []net.IP
  22         Port int
  23 }
  24
  25 // mDNSProtocol mdns protocol interface.
  26 type mDNSProtocol interface {
  27         registerService(port int) error
  28         registerResolver(event chan LANPeerEvent) error
  29         stopService()
  30         stopResolver()
  31 }
  32
  33 // LANDiscover responsible for finding the related services registered LAN nodes.
  34 type LANDiscover struct {
  35         protocol        mDNSProtocol
  36         resolving       uint32
  37         servicePort     int //service port
  38         entries         chan LANPeerEvent
  39         eventDispatcher *event.Dispatcher
  40         quite           chan struct{}
  41 }
  42
  43 // NewLANDiscover create a new LAN node discover.
  44 func NewLANDiscover(protocol mDNSProtocol, port int) *LANDiscover {
  45         ld := &LANDiscover{
  46                 protocol:        protocol,
  47                 servicePort:     port,
  48                 entries:         make(chan LANPeerEvent, 1024),
  49                 eventDispatcher: event.NewDispatcher(),
  50                 quite:           make(chan struct{}),
  51         }
  52         // register service
  53         go ld.registerServiceRoutine()
  54         go ld.getLANPeerLoop()
  55         return ld
  56 }
  57
  58 // Stop stop LAN discover.
  59 func (ld *LANDiscover) Stop() {
  60         close(ld.quite)
  61         ld.protocol.stopResolver()
  62         ld.protocol.stopService()
  63         ld.eventDispatcher.Stop()
  64 }
  65
  66 // Subscribe used to subscribe for LANPeerEvent.
  67 func (ld *LANDiscover) Subscribe() (*event.Subscription, error) {
  68         //subscribe LANPeerEvent.
  69         sub, err := ld.eventDispatcher.Subscribe(LANPeerEvent{})
  70         if err != nil {
  71                 return nil, err
  72         }
  73
  74         //need to register the parser once.
  75         if atomic.CompareAndSwapUint32(&ld.resolving, 0, 1) {
  76                 if err = ld.protocol.registerResolver(ld.entries); err != nil {
  77                         return nil, err
  78                 }
  79         }
  80
  81         return sub, nil
  82 }
  83
  84 // register service routine, will be re-registered periodically
  85 // for the stability of node discovery.
  86 func (ld *LANDiscover) registerServiceRoutine() {
  87         time.Sleep(registerServiceDelay)
  88         if err := ld.protocol.registerService(ld.servicePort); err != nil {
  89                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("mdns service register error")
  90                 return
  91         }
  92
  93         ticker := time.NewTicker(registerServiceCycle)
  94         defer ticker.Stop()
  95         for {
  96                 select {
  97                 case <-ticker.C:
  98                         ld.protocol.stopService()
  99                         if err := ld.protocol.registerService(ld.servicePort); err != nil {
 100                                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("mdns service register error")
 101                                 return
 102                         }
 103                 case <-ld.quite:
 104                         return
 105                 }
 106         }
 107 }
 108
 109 // obtain the lan peer event from the specific protocol
 110 // and distribute it to the subscriber.
 111 func (ld *LANDiscover) getLANPeerLoop() {
 112         for {
 113                 select {
 114                 case entry := <-ld.entries:
 115                         if err := ld.eventDispatcher.Post(entry); err != nil {
 116                                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("event dispatch error")
 117                         }
 118                 case <-ld.quite:
 119                         return
 120                 }
 121         }
 122 }