implement basic peer discovery using mdns
Jeromy committed
Apr 21, 2015 at 17:24 UTC
0917c1cb82deb5748949af6c359f28ff9df448ac
2 files changed
+194
core/core.go
+19
@@ -18,6 +18,7 @@ import (
18
19
diag "github.com/ipfs/go-ipfs/diagnostics"
20
ic "github.com/ipfs/go-ipfs/p2p/crypto"
21
+ discovery "github.com/ipfs/go-ipfs/p2p/discovery"
22
p2phost "github.com/ipfs/go-ipfs/p2p/host"
23
p2pbhost "github.com/ipfs/go-ipfs/p2p/host/basic"
24
rhost "github.com/ipfs/go-ipfs/p2p/host/routed"
@@ -83,6 +84,7 @@ type IpfsNode struct {
84
DAG merkledag.DAGService // the merkle dag service, get/add objects.
85
Resolver *path.Resolver // the path resolution system
86
Reporter metrics.Reporter
87
+ Discovery discovery.Service
88
89
// Online
90
PeerHost p2phost.Host // the network host (server+client)
@@ -261,9 +263,26 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
263
n.Reprovider = rp.NewReprovider(n.Routing, n.Blockstore)
264
go n.Reprovider.ProvideEvery(ctx, kReprovideFrequency)
265
266
+ // setup local discovery
267
+ service, err := discovery.NewMdnsService(n.PeerHost)
268
+ if err != nil {
269
+ return err
270
+ }
271
+ service.RegisterNotifee(n)
272
+ n.Discovery = service
273
+
274
return n.Bootstrap(DefaultBootstrapConfig)
275
}
276
277
+func (n *IpfsNode) HandlePeerFound(p peer.PeerInfo) {
278
+ log.Warning("trying peer info: ", p)
279
+ ctx, _ := context.WithTimeout(n.Context(), time.Second*10)
280
+ err := n.PeerHost.Connect(ctx, p)
281
+ if err != nil {
282
+ log.Warning("Failed to connect to peer found by discovery: ", err)
283
+ }
284
+}
285
+
286
// startOnlineServicesWithHost is the set of services which need to be
287
// initialized with the host and _before_ we start listening.
288
func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost.Host, routingOption RoutingOption) error {
p2p/discovery/mdns.go
new
+175
@@ -0,0 +1,175 @@
1
+package discovery
2
+
3
+import (
4
+ "fmt"
5
+ "io"
6
+ "io/ioutil"
7
+ golog "log"
8
+ "net"
9
+ "strconv"
10
+ "strings"
11
+ "sync"
12
+ "time"
13
+
14
+ "github.com/hashicorp/mdns"
15
+ ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
16
+ manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
17
+
18
+ "github.com/ipfs/go-ipfs/p2p/host"
19
+ "github.com/ipfs/go-ipfs/p2p/peer"
20
+ u "github.com/ipfs/go-ipfs/util"
21
+)
22
+
23
+var log = u.Logger("mdns")
24
+
25
+const LookupFrequency = time.Second * 5
26
+const ServiceTag = "discovery.ipfs.io"
27
+
28
+type Service interface {
29
+ io.Closer
30
+ RegisterNotifee(Notifee)
31
+ UnregisterNotifee(Notifee)
32
+}
33
+
34
+type Notifee interface {
35
+ HandlePeerFound(peer.PeerInfo)
36
+}
37
+
38
+type mdnsService struct {
39
+ server *mdns.Server
40
+ service *mdns.MDNSService
41
+ host host.Host
42
+
43
+ lk sync.Mutex
44
+ notifees []Notifee
45
+}
46
+
47
+func NewMdnsService(peerhost host.Host) (Service, error) {
48
+
49
+ // TODO: dont let mdns use logging...
50
+ golog.SetOutput(ioutil.Discard)
51
+
52
+ // determine my local swarm port
53
+ port := 4001
54
+ for _, addr := range peerhost.Addrs() {
55
+ parts := strings.Split(addr.String(), "/")
56
+ fmt.Println("parts len: ", len(parts))
57
+ if len(parts) == 5 && parts[3] == "tcp" {
58
+ n, err := strconv.Atoi(parts[4])
59
+ if err != nil {
60
+ return nil, err
61
+ }
62
+ port = n
63
+ break
64
+ }
65
+ }
66
+ fmt.Println("using port: ", port)
67
+
68
+ myid := peerhost.ID().Pretty()
69
+
70
+ info := []string{myid}
71
+ service, err := mdns.NewMDNSService(myid, ServiceTag, "", "", port, nil, info)
72
+ if err != nil {
73
+ return nil, err
74
+ }
75
+
76
+ // Create the mDNS server, defer shutdown
77
+ server, err := mdns.NewServer(&mdns.Config{Zone: service})
78
+ if err != nil {
79
+ return nil, err
80
+ }
81
+
82
+ s := &mdnsService{
83
+ server: server,
84
+ service: service,
85
+ host: peerhost,
86
+ }
87
+
88
+ go s.pollForEntries()
89
+
90
+ return s, nil
91
+}
92
+
93
+func (m *mdnsService) Close() error {
94
+ return m.server.Shutdown()
95
+}
96
+
97
+func (m *mdnsService) pollForEntries() {
98
+ ticker := time.NewTicker(LookupFrequency)
99
+ for {
100
+ select {
101
+ case <-ticker.C:
102
+ entriesCh := make(chan *mdns.ServiceEntry, 16)
103
+ go func() {
104
+ for entry := range entriesCh {
105
+ m.handleEntry(entry)
106
+ }
107
+ }()
108
+
109
+ qp := mdns.QueryParam{}
110
+ qp.Domain = "local"
111
+ qp.Entries = entriesCh
112
+ qp.Service = ServiceTag
113
+ qp.Timeout = time.Second * 3
114
+
115
+ err := mdns.Query(&qp)
116
+ if err != nil {
117
+ log.Error("mdns lookup error: ", err)
118
+ }
119
+ close(entriesCh)
120
+ }
121
+ }
122
+}
123
+
124
+func (m *mdnsService) handleEntry(e *mdns.ServiceEntry) {
125
+ mpeer, err := peer.IDB58Decode(e.Info)
126
+ if err != nil {
127
+ log.Warning("Error parsing peer ID from mdns entry: ", err)
128
+ return
129
+ }
130
+
131
+ if mpeer == m.host.ID() {
132
+ return
133
+ }
134
+
135
+ maddr, err := manet.FromNetAddr(&net.TCPAddr{
136
+ IP: e.AddrV4,
137
+ Port: e.Port,
138
+ })
139
+ if err != nil {
140
+ log.Warning("Error parsing multiaddr from mdns entry: ", err)
141
+ return
142
+ }
143
+
144
+ pi := peer.PeerInfo{
145
+ ID: mpeer,
146
+ Addrs: []ma.Multiaddr{maddr},
147
+ }
148
+
149
+ m.lk.Lock()
150
+ for _, n := range m.notifees {
151
+ n.HandlePeerFound(pi)
152
+ }
153
+ m.lk.Unlock()
154
+}
155
+
156
+func (m *mdnsService) RegisterNotifee(n Notifee) {
157
+ m.lk.Lock()
158
+ m.notifees = append(m.notifees, n)
159
+ m.lk.Unlock()
160
+}
161
+
162
+func (m *mdnsService) UnregisterNotifee(n Notifee) {
163
+ m.lk.Lock()
164
+ found := -1
165
+ for i, notif := range m.notifees {
166
+ if notif == n {
167
+ found = i
168
+ break
169
+ }
170
+ }
171
+ if found != -1 {
172
+ m.notifees = append(m.notifees[:found], m.notifees[found+1:]...)
173
+ }
174
+ m.lk.Unlock()
175
+}