OSDN Git Service

writer close
[bytom/vapor.git] / p2p / listener.go
1 package p2p
2
3 import (
4         "fmt"
5         "net"
6         "strconv"
7         "strings"
8         "time"
9
10         log "github.com/sirupsen/logrus"
11         cmn "github.com/tendermint/tmlibs/common"
12
13         cfg "github.com/vapor/config"
14         "github.com/vapor/errors"
15         "github.com/vapor/p2p/upnp"
16 )
17
18 const (
19         numBufferedConnections = 10
20         defaultExternalPort    = 8770
21         tryListenTimes         = 5
22 )
23
24 //Listener subset of the methods of DefaultListener
25 type Listener interface {
26         Connections() <-chan net.Conn
27         InternalAddress() *NetAddress
28         ExternalAddress() *NetAddress
29         String() string
30         Stop() bool
31 }
32
33 // Defaults to tcp
34 func protocolAndAddress(listenAddr string) (string, string) {
35         p, address := "tcp", listenAddr
36         parts := strings.SplitN(address, "://", 2)
37         if len(parts) == 2 {
38                 p, address = parts[0], parts[1]
39         }
40         return p, address
41 }
42
43 // GetListener get listener and listen address.
44 func GetListener(config *cfg.P2PConfig) (Listener, string) {
45         p, address := protocolAndAddress(config.ListenAddress)
46         l, listenerStatus := NewDefaultListener(p, address, config.SkipUPNP)
47
48         // We assume that the rpcListener has the same ExternalAddress.
49         // This is probably true because both P2P and RPC listeners use UPnP,
50         // except of course if the rpc is only bound to localhost
51         if listenerStatus {
52                 return l, cmn.Fmt("%v:%v", l.ExternalAddress().IP.String(), l.ExternalAddress().Port)
53         }
54
55         return l, cmn.Fmt("%v:%v", l.InternalAddress().IP.String(), l.InternalAddress().Port)
56 }
57
58 //getUPNPExternalAddress UPNP external address discovery & port mapping
59 func getUPNPExternalAddress(externalPort, internalPort int) (*NetAddress, error) {
60         nat, err := upnp.Discover()
61         if err != nil {
62                 return nil, errors.Wrap(err, "could not perform UPNP discover")
63         }
64
65         ext, err := nat.GetExternalAddress()
66         if err != nil {
67                 return nil, errors.Wrap(err, "could not perform UPNP external address")
68         }
69
70         if externalPort == 0 {
71                 externalPort = defaultExternalPort
72         }
73         externalPort, err = nat.AddPortMapping("tcp", externalPort, internalPort, "vapord tcp", 0)
74         if err != nil {
75                 return nil, errors.Wrap(err, "could not add tcp UPNP port mapping")
76         }
77         externalPort, err = nat.AddPortMapping("udp", externalPort, internalPort, "vapord udp", 0)
78         if err != nil {
79                 return nil, errors.Wrap(err, "could not add udp UPNP port mapping")
80         }
81         return NewNetAddressIPPort(ext, uint16(externalPort)), nil
82 }
83
84 func splitHostPort(addr string) (host string, port int) {
85         host, portStr, err := net.SplitHostPort(addr)
86         if err != nil {
87                 cmn.PanicSanity(err)
88         }
89         port, err = strconv.Atoi(portStr)
90         if err != nil {
91                 cmn.PanicSanity(err)
92         }
93         return host, port
94 }
95
96 //DefaultListener Implements vapord server Listener
97 type DefaultListener struct {
98         cmn.BaseService
99
100         listener    net.Listener
101         intAddr     *NetAddress
102         extAddr     *NetAddress
103         connections chan net.Conn
104 }
105
106 //NewDefaultListener create a default listener
107 func NewDefaultListener(protocol string, lAddr string, skipUPNP bool) (Listener, bool) {
108         // Local listen IP & port
109         lAddrIP, lAddrPort := splitHostPort(lAddr)
110
111         listener, err := net.Listen(protocol, lAddr)
112         for i := 0; i < tryListenTimes && err != nil; i++ {
113                 time.Sleep(time.Second * 1)
114                 listener, err = net.Listen(protocol, lAddr)
115         }
116         if err != nil {
117                 log.Panic(err)
118         }
119
120         intAddr, err := NewNetAddressString(lAddr)
121         if err != nil {
122                 log.Panic(err)
123         }
124
125         // Actual listener local IP & port
126         listenerIP, listenerPort := splitHostPort(listener.Addr().String())
127         log.Info("Local listener", " ip:", listenerIP, " port:", listenerPort)
128
129         // Determine external address...
130         var extAddr *NetAddress
131         var upnpMap bool
132         if !skipUPNP && (lAddrIP == "" || lAddrIP == "0.0.0.0") {
133                 extAddr, err = getUPNPExternalAddress(lAddrPort, listenerPort)
134                 upnpMap = err == nil
135                 log.WithFields(log.Fields{"module": logModule, "err": err}).Info("get UPNP external address")
136         }
137
138         // Get the IPv4 available
139         if extAddr == nil {
140                 if ip, err := ExternalIPv4(); err != nil {
141                         log.WithFields(log.Fields{"module": logModule, "err": err}).Warning("get ipv4 external address")
142                         log.Panic("get ipv4 external address fail!")
143                 } else {
144                         extAddr = NewNetAddressIPPort(net.ParseIP(ip), uint16(lAddrPort))
145                         log.WithFields(log.Fields{"module": logModule, "addr": extAddr}).Info("get ipv4 external address success")
146                 }
147         }
148
149         dl := &DefaultListener{
150                 listener:    listener,
151                 intAddr:     intAddr,
152                 extAddr:     extAddr,
153                 connections: make(chan net.Conn, numBufferedConnections),
154         }
155         dl.BaseService = *cmn.NewBaseService(nil, "DefaultListener", dl)
156         dl.Start() // Started upon construction
157         if upnpMap {
158                 return dl, true
159         }
160
161         conn, err := net.DialTimeout("tcp", extAddr.String(), 3*time.Second)
162         if err != nil {
163                 return dl, false
164         }
165         conn.Close()
166         return dl, true
167 }
168
169 //OnStart start listener
170 func (l *DefaultListener) OnStart() error {
171         l.BaseService.OnStart()
172         go l.listenRoutine()
173         return nil
174 }
175
176 //OnStop stop listener
177 func (l *DefaultListener) OnStop() {
178         l.BaseService.OnStop()
179         l.listener.Close()
180 }
181
182 //listenRoutine Accept connections and pass on the channel
183 func (l *DefaultListener) listenRoutine() {
184         for {
185                 conn, err := l.listener.Accept()
186                 if !l.IsRunning() {
187                         break // Go to cleanup
188                 }
189                 // listener wasn't stopped,
190                 // yet we encountered an error.
191                 if err != nil {
192                         log.Panic(err)
193                 }
194                 l.connections <- conn
195         }
196         // Cleanup
197         close(l.connections)
198 }
199
200 //Connections a channel of inbound connections. It gets closed when the listener closes.
201 func (l *DefaultListener) Connections() <-chan net.Conn {
202         return l.connections
203 }
204
205 //InternalAddress listener internal address
206 func (l *DefaultListener) InternalAddress() *NetAddress {
207         return l.intAddr
208 }
209
210 //ExternalAddress listener external address for remote peer dial
211 func (l *DefaultListener) ExternalAddress() *NetAddress {
212         return l.extAddr
213 }
214
215 // NetListener the returned listener is already Accept()'ing. So it's not suitable to pass into http.Serve().
216 func (l *DefaultListener) NetListener() net.Listener {
217         return l.listener
218 }
219
220 //String string of default listener
221 func (l *DefaultListener) String() string {
222         return fmt.Sprintf("Listener(@%v)", l.extAddr)
223 }