OSDN Git Service

try to remove RepeatTimer for fix dead lock
[bytom/bytom.git] / p2p / listener.go
1 package p2p
2
3 import (
4         "fmt"
5         "net"
6         "strconv"
7         "time"
8
9         "github.com/bytom/p2p/upnp"
10         log "github.com/sirupsen/logrus"
11         cmn "github.com/tendermint/tmlibs/common"
12         tlog "github.com/tendermint/tmlibs/log"
13 )
14
15 type Listener interface {
16         Connections() <-chan net.Conn
17         InternalAddress() *NetAddress
18         ExternalAddress() *NetAddress
19         String() string
20         Stop() bool
21 }
22
23 // Implements Listener
24 type DefaultListener struct {
25         cmn.BaseService
26
27         listener    net.Listener
28         intAddr     *NetAddress
29         extAddr     *NetAddress
30         connections chan net.Conn
31 }
32
33 const (
34         numBufferedConnections = 10
35         defaultExternalPort    = 8770
36         tryListenSeconds       = 5
37 )
38
39 func splitHostPort(addr string) (host string, port int) {
40         host, portStr, err := net.SplitHostPort(addr)
41         if err != nil {
42                 cmn.PanicSanity(err)
43         }
44         port, err = strconv.Atoi(portStr)
45         if err != nil {
46                 cmn.PanicSanity(err)
47         }
48         return host, port
49 }
50
51 // skipUPNP: If true, does not try getUPNPExternalAddress()
52 func NewDefaultListener(protocol string, lAddr string, skipUPNP bool, logger tlog.Logger) Listener {
53         // Local listen IP & port
54         lAddrIP, lAddrPort := splitHostPort(lAddr)
55
56         // Create listener
57         var listener net.Listener
58         var err error
59         for i := 0; i < tryListenSeconds; i++ {
60                 listener, err = net.Listen(protocol, lAddr)
61                 if err == nil {
62                         break
63                 } else if i < tryListenSeconds-1 {
64                         time.Sleep(time.Second * 1)
65                 }
66         }
67         if err != nil {
68                 cmn.PanicCrisis(err)
69         }
70         // Actual listener local IP & port
71         listenerIP, listenerPort := splitHostPort(listener.Addr().String())
72         log.WithFields(log.Fields{
73                 "ip":   listenerIP,
74                 "port": listenerPort,
75         }).Info("Local listener")
76
77         // Determine internal address...
78         var intAddr *NetAddress
79         intAddr, err = NewNetAddressString(lAddr)
80         if err != nil {
81                 cmn.PanicCrisis(err)
82         }
83
84         // Determine external address...
85         var extAddr *NetAddress
86         if !skipUPNP {
87                 // If the lAddrIP is INADDR_ANY, try UPnP
88                 if lAddrIP == "" || lAddrIP == "0.0.0.0" {
89                         extAddr = getUPNPExternalAddress(lAddrPort, listenerPort)
90                 }
91         }
92         // Otherwise just use the local address...
93         if extAddr == nil {
94                 extAddr = getNaiveExternalAddress(listenerPort)
95         }
96         if extAddr == nil {
97                 cmn.PanicCrisis("Could not determine external address!")
98         }
99
100         dl := &DefaultListener{
101                 listener:    listener,
102                 intAddr:     intAddr,
103                 extAddr:     extAddr,
104                 connections: make(chan net.Conn, numBufferedConnections),
105         }
106         dl.BaseService = *cmn.NewBaseService(logger, "DefaultListener", dl)
107         dl.Start() // Started upon construction
108         return dl
109 }
110
111 func (l *DefaultListener) OnStart() error {
112         l.BaseService.OnStart()
113         go l.listenRoutine()
114         return nil
115 }
116
117 func (l *DefaultListener) OnStop() {
118         l.BaseService.OnStop()
119         l.listener.Close()
120 }
121
122 // Accept connections and pass on the channel
123 func (l *DefaultListener) listenRoutine() {
124         for {
125                 conn, err := l.listener.Accept()
126
127                 if !l.IsRunning() {
128                         break // Go to cleanup
129                 }
130
131                 // listener wasn't stopped,
132                 // yet we encountered an error.
133                 if err != nil {
134                         cmn.PanicCrisis(err)
135                 }
136
137                 l.connections <- conn
138         }
139
140         // Cleanup
141         close(l.connections)
142         for _ = range l.connections {
143                 // Drain
144         }
145 }
146
147 // A channel of inbound connections.
148 // It gets closed when the listener closes.
149 func (l *DefaultListener) Connections() <-chan net.Conn {
150         return l.connections
151 }
152
153 func (l *DefaultListener) InternalAddress() *NetAddress {
154         return l.intAddr
155 }
156
157 func (l *DefaultListener) ExternalAddress() *NetAddress {
158         return l.extAddr
159 }
160
161 // NOTE: The returned listener is already Accept()'ing.
162 // So it's not suitable to pass into http.Serve().
163 func (l *DefaultListener) NetListener() net.Listener {
164         return l.listener
165 }
166
167 func (l *DefaultListener) String() string {
168         return fmt.Sprintf("Listener(@%v)", l.extAddr)
169 }
170
171 /* external address helpers */
172
173 // UPNP external address discovery & port mapping
174 func getUPNPExternalAddress(externalPort, internalPort int) *NetAddress {
175         log.Info("Getting UPNP external address")
176         nat, err := upnp.Discover()
177         if err != nil {
178                 log.WithField("error", err).Error("Could not perform UPNP discover")
179                 return nil
180         }
181
182         ext, err := nat.GetExternalAddress()
183         if err != nil {
184                 log.WithField("error", err).Error("Could not perform UPNP external address")
185                 return nil
186         }
187
188         // UPnP can't seem to get the external port, so let's just be explicit.
189         if externalPort == 0 {
190                 externalPort = defaultExternalPort
191         }
192
193         externalPort, err = nat.AddPortMapping("tcp", externalPort, internalPort, "bytomd", 0)
194         if err != nil {
195                 log.WithField("error", err).Error("Could not add UPNP port mapping")
196                 return nil
197         }
198
199         log.WithField("address", ext).Info("Got UPNP external address")
200         return NewNetAddressIPPort(ext, uint16(externalPort))
201 }
202
203 // TODO: use syscalls: http://pastebin.com/9exZG4rh
204 func getNaiveExternalAddress(port int) *NetAddress {
205         addrs, err := net.InterfaceAddrs()
206         if err != nil {
207                 cmn.PanicCrisis(cmn.Fmt("Could not fetch interface addresses: %v", err))
208         }
209
210         for _, a := range addrs {
211                 ipnet, ok := a.(*net.IPNet)
212                 if !ok {
213                         continue
214                 }
215                 v4 := ipnet.IP.To4()
216                 if v4 == nil || v4[0] == 127 {
217                         continue
218                 } // loopback
219                 return NewNetAddressIPPort(ipnet.IP, uint16(port))
220         }
221         return nil
222 }