making connections between nodes get closer to working
Jeromy committed
Jul 31, 2014 at 21:55 UTC
8d98d4b48dd4d47cf7a4c0aca9e9c192db64617a
4 files changed
+56
-34
identify/identify.go
+6
-8
@@ -4,12 +4,12 @@ package identify
4
5
import (
6
peer "github.com/jbenet/go-ipfs/peer"
7
- swarm "github.com/jbenet/go-ipfs/swarm"
7
+ u "github.com/jbenet/go-ipfs/util"
8
)
9
10
// Perform initial communication with this peer to share node ID's and
11
// initiate communication
12
-func Handshake(self *peer.Peer, conn *swarm.Conn) error {
12
+func Handshake(self, remote *peer.Peer, in, out chan []byte) error {
13
14
// temporary:
15
// put your own id in a 16byte buffer and send that over to
@@ -17,12 +17,10 @@ func Handshake(self *peer.Peer, conn *swarm.Conn) error {
17
// Once that trade is finished, the handshake is complete and
18
// both sides should 'trust' each other
19
20
- id := make([]byte, 16)
21
- copy(id, self.ID)
22
-
23
- conn.Outgoing.MsgChan <- id
24
- resp := <-conn.Incoming.MsgChan
25
- conn.Peer.ID = peer.ID(resp)
20
+ out <- self.ID
21
+ resp := <-in
22
+ remote.ID = peer.ID(resp)
23
+ u.DOut("Got node id: %s", string(remote.ID))
24
25
return nil
26
}
routing/dht/dht.go
+25
-7
@@ -55,6 +55,10 @@ func NewDHT(p *peer.Peer) (*IpfsDHT, error) {
55
return dht, nil
56
}
57
58
+func (dht *IpfsDHT) Start() {
59
+ go dht.handleMessages()
60
+}
61
+
62
// Connect to a new peer at the given address
63
func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) error {
64
peer := new(peer.Peer)
@@ -65,24 +69,26 @@ func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) error {
69
return err
70
}
71
68
- err = identify.Handshake(dht.self, conn)
72
+ err = identify.Handshake(dht.self, peer, conn.Incoming.MsgChan, conn.Outgoing.MsgChan)
73
if err != nil {
74
return err
75
}
76
73
- dht.network.StartConn(conn.Peer.Key(), conn)
77
+ dht.network.StartConn(conn)
78
79
// TODO: Add this peer to our routing table
80
return nil
81
}
82
79
-
83
// Read in all messages from swarm and handle them appropriately
84
// NOTE: this function is just a quick sketch
85
func (dht *IpfsDHT) handleMessages() {
86
+ u.DOut("Being message handling routine")
87
for {
88
select {
89
case mes := <-dht.network.Chan.Incoming:
90
+ u.DOut("recieved message from swarm.")
91
+
92
pmes := new(DHTMessage)
93
err := proto.Unmarshal(mes.Data, pmes)
94
if err != nil {
@@ -118,6 +124,8 @@ func (dht *IpfsDHT) handleMessages() {
124
dht.handleFindNode(mes.Peer, pmes)
125
}
126
127
+ case err := <-dht.network.Chan.Errors:
128
+ panic(err)
129
case <-dht.shutdown:
130
return
131
}
@@ -158,10 +166,6 @@ func (dht *IpfsDHT) handlePutValue(p *peer.Peer, pmes *DHTMessage) {
166
}
167
}
168
161
-func (dht *IpfsDHT) handleFindNode(p *peer.Peer, pmes *DHTMessage) {
162
- panic("Not implemented.")
163
-}
164
-
169
func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *DHTMessage) {
170
isResponse := true
171
resp := new(DHTMessage)
@@ -172,6 +176,18 @@ func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *DHTMessage) {
176
dht.network.Chan.Outgoing <-swarm.NewMessage(p, []byte(resp.String()))
177
}
178
179
+func (dht *IpfsDHT) handleFindNode(p *peer.Peer, pmes *DHTMessage) {
180
+ panic("Not implemented.")
181
+}
182
+
183
+func (dht *IpfsDHT) handleGetProviders(p *peer.Peer, pmes *DHTMessage) {
184
+ panic("Not implemented.")
185
+}
186
+
187
+func (dht *IpfsDHT) handleAddProvider(p *peer.Peer, pmes *DHTMessage) {
188
+ panic("Not implemented.")
189
+}
190
+
191
192
// Register a handler for a specific message ID, used for getting replies
193
// to certain messages (i.e. response to a GET_VALUE message)
@@ -202,6 +218,8 @@ func (dht *IpfsDHT) Halt() {
218
// Ping a node, log the time it took
219
func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) {
220
// Thoughts: maybe this should accept an ID and do a peer lookup?
221
+ u.DOut("Enter Ping.")
222
+
223
id := GenerateMessageID()
224
mes_type := DHTMessage_PING
225
pmes := new(DHTMessage)
swarm/swarm.go
+23
-17
@@ -8,6 +8,7 @@ import (
8
peer "github.com/jbenet/go-ipfs/peer"
9
u "github.com/jbenet/go-ipfs/util"
10
ma "github.com/jbenet/go-multiaddr"
11
+ ident "github.com/jbenet/go-ipfs/identify"
12
)
13
14
// Message represents a packet of information sent to or received from a
@@ -109,27 +110,21 @@ func (s *Swarm) connListen(maddr *ma.Multiaddr) error {
110
111
// Handle getting ID from this peer and adding it into the map
112
func (s *Swarm) handleNewConn(nconn net.Conn) {
112
- p := MakePeerFromConn(nconn)
113
-
114
- var addr *ma.Multiaddr
115
-
116
- //naddr := nconn.RemoteAddr()
117
- //addr := ma.FromDialArgs(naddr.Network(), naddr.String())
113
+ p := new(peer.Peer)
114
115
conn := &Conn{
116
Peer: p,
121
- Addr: addr,
117
+ Addr: nil,
118
Conn: nconn,
119
}
124
-
120
newConnChans(conn)
126
- go s.fanIn(conn)
127
-}
121
129
-// Negotiate with peer for its ID and create a peer object
130
-// TODO: this might belong in the peer package
131
-func MakePeerFromConn(conn net.Conn) *peer.Peer {
132
- panic("Not yet implemented.")
122
+ err := ident.Handshake(s.local, p, conn.Incoming.MsgChan, conn.Outgoing.MsgChan)
123
+ if err != nil {
124
+ panic(err)
125
+ }
126
+
127
+ s.StartConn(conn)
128
}
129
130
// Close closes a swarm.
@@ -170,14 +165,19 @@ func (s *Swarm) Dial(peer *peer.Peer) (*Conn, error) {
165
return nil, err
166
}
167
173
- s.StartConn(k, conn)
168
+ s.StartConn(conn)
169
return conn, nil
170
}
171
177
-func (s *Swarm) StartConn(k u.Key, conn *Conn) {
172
+func (s *Swarm) StartConn(conn *Conn) {
173
+ if conn == nil {
174
+ panic("tried to start nil Conn!")
175
+ }
176
+
177
+ u.DOut("Starting connection: %s", string(conn.Peer.ID))
178
// add to conns
179
s.connsLock.Lock()
180
- s.conns[k] = conn
180
+ s.conns[conn.Peer.Key()] = conn
181
s.connsLock.Unlock()
182
183
// kick off reader goroutine
@@ -191,6 +191,7 @@ func (s *Swarm) fanOut() {
191
case <-s.Chan.Close:
192
return // told to close.
193
case msg, ok := <-s.Chan.Outgoing:
194
+ u.DOut("fanOut: outgoing message for: '%s'", msg.Peer.Key())
195
if !ok {
196
return
197
}
@@ -198,14 +199,17 @@ func (s *Swarm) fanOut() {
199
s.connsLock.RLock()
200
conn, found := s.conns[msg.Peer.Key()]
201
s.connsLock.RUnlock()
202
+
203
if !found {
204
e := fmt.Errorf("Sent msg to peer without open conn: %v",
205
msg.Peer)
206
s.Chan.Errors <- e
207
+ continue
208
}
209
210
// queue it in the connection's buffer
211
conn.Outgoing.MsgChan <- msg.Data
212
+ u.DOut("fanOut: message off.")
213
}
214
}
215
}
@@ -225,6 +229,7 @@ Loop:
229
break Loop
230
231
case data, ok := <-conn.Incoming.MsgChan:
232
+ u.DOut("fanIn: got message from incoming channel.")
233
if !ok {
234
e := fmt.Errorf("Error retrieving from conn: %v", conn)
235
s.Chan.Errors <- e
@@ -234,6 +239,7 @@ Loop:
239
// wrap it for consumers.
240
msg := &Message{Peer: conn.Peer, Data: data}
241
s.Chan.Incoming <- msg
242
+ u.DOut("fanIn: message off.")
243
}
244
}
245
util/util.go
+2
-2
@@ -41,12 +41,12 @@ func TildeExpansion(filename string) (string, error) {
41
42
// PErr is a shorthand printing function to output to Stderr.
43
func PErr(format string, a ...interface{}) {
44
- fmt.Fprintf(os.Stderr, format, a...)
44
+ fmt.Fprintf(os.Stderr, format + "\n", a...)
45
}
46
47
// POut is a shorthand printing function to output to Stdout.
48
func POut(format string, a ...interface{}) {
49
- fmt.Fprintf(os.Stdout, format, a...)
49
+ fmt.Fprintf(os.Stdout, format + "\n", a...)
50
}
51
52
// DErr is a shorthand debug printing function to output to Stderr.